mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-12-17 03:42:48 +00:00
92 lines
3.3 KiB
C++
92 lines
3.3 KiB
C++
#include <Storages/System/StorageSystemClusters.h>
|
|
#include <Interpreters/Cluster.h>
|
|
#include <Columns/ColumnString.h>
|
|
#include <Columns/ColumnsNumber.h>
|
|
#include <DataTypes/DataTypeString.h>
|
|
#include <DataTypes/DataTypesNumber.h>
|
|
#include <DataStreams/OneBlockInputStream.h>
|
|
#include <Interpreters/Context.h>
|
|
|
|
namespace DB
|
|
{
|
|
|
|
|
|
StorageSystemClusters::StorageSystemClusters(const std::string & name_)
|
|
: name(name_)
|
|
{
|
|
columns = ColumnsDescription({
|
|
{ "cluster", std::make_shared<DataTypeString>() },
|
|
{ "shard_num", std::make_shared<DataTypeUInt32>() },
|
|
{ "shard_weight", std::make_shared<DataTypeUInt32>() },
|
|
{ "replica_num", std::make_shared<DataTypeUInt32>() },
|
|
{ "host_name", std::make_shared<DataTypeString>() },
|
|
{ "host_address", std::make_shared<DataTypeString>() },
|
|
{ "port", std::make_shared<DataTypeUInt16>() },
|
|
{ "is_local", std::make_shared<DataTypeUInt8>() },
|
|
{ "user", std::make_shared<DataTypeString>() },
|
|
{ "default_database", std::make_shared<DataTypeString>() },
|
|
});
|
|
}
|
|
|
|
|
|
BlockInputStreams StorageSystemClusters::read(
|
|
const Names & column_names,
|
|
const SelectQueryInfo &,
|
|
const Context & context,
|
|
QueryProcessingStage::Enum & processed_stage,
|
|
const size_t /*max_block_size*/,
|
|
const unsigned /*num_streams*/)
|
|
{
|
|
check(column_names);
|
|
processed_stage = QueryProcessingStage::FetchColumns;
|
|
|
|
MutableColumns res_columns = getSampleBlock().cloneEmptyColumns();
|
|
|
|
auto updateColumns = [&](const std::string & cluster_name, const Cluster::ShardInfo & shard_info,
|
|
const Cluster::Address & address)
|
|
{
|
|
size_t i = 0;
|
|
res_columns[i++]->insert(cluster_name);
|
|
res_columns[i++]->insert(static_cast<UInt64>(shard_info.shard_num));
|
|
res_columns[i++]->insert(static_cast<UInt64>(shard_info.weight));
|
|
res_columns[i++]->insert(static_cast<UInt64>(address.replica_num));
|
|
res_columns[i++]->insert(address.host_name);
|
|
res_columns[i++]->insert(address.resolved_address.host().toString());
|
|
res_columns[i++]->insert(static_cast<UInt64>(address.port));
|
|
res_columns[i++]->insert(static_cast<UInt64>(shard_info.isLocal()));
|
|
res_columns[i++]->insert(address.user);
|
|
res_columns[i++]->insert(address.default_database);
|
|
};
|
|
|
|
auto clusters = context.getClusters().getContainer();
|
|
for (const auto & entry : clusters)
|
|
{
|
|
const std::string cluster_name = entry.first;
|
|
const ClusterPtr cluster = entry.second;
|
|
const auto & addresses_with_failover = cluster->getShardsAddresses();
|
|
const auto & shards_info = cluster->getShardsInfo();
|
|
|
|
if (!addresses_with_failover.empty())
|
|
{
|
|
auto it1 = addresses_with_failover.cbegin();
|
|
auto it2 = shards_info.cbegin();
|
|
|
|
while (it1 != addresses_with_failover.cend())
|
|
{
|
|
const auto & addresses = *it1;
|
|
const auto & shard_info = *it2;
|
|
|
|
for (const auto & address : addresses)
|
|
updateColumns(cluster_name, shard_info, address);
|
|
|
|
++it1;
|
|
++it2;
|
|
}
|
|
}
|
|
}
|
|
|
|
return BlockInputStreams(1, std::make_shared<OneBlockInputStream>(getSampleBlock().cloneWithColumns(std::move(res_columns))));
|
|
}
|
|
|
|
}
|