2017-04-01 09:19:00 +00:00
|
|
|
#include <Columns/ColumnString.h>
|
|
|
|
#include <DataTypes/DataTypeString.h>
|
|
|
|
#include <DataTypes/DataTypesNumber.h>
|
|
|
|
#include <DataTypes/DataTypeDateTime.h>
|
|
|
|
#include <DataStreams/OneBlockInputStream.h>
|
|
|
|
#include <Storages/System/StorageSystemReplicas.h>
|
|
|
|
#include <Storages/StorageReplicatedMergeTree.h>
|
2017-11-20 05:22:54 +00:00
|
|
|
#include <Storages/VirtualColumnUtils.h>
|
2017-07-13 20:58:19 +00:00
|
|
|
#include <Common/typeid_cast.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Databases/IDatabase.h>
|
2014-10-06 05:18:17 +00:00
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
|
2015-01-21 03:56:28 +00:00
|
|
|
StorageSystemReplicas::StorageSystemReplicas(const std::string & name_)
|
2017-04-01 07:20:54 +00:00
|
|
|
: name(name_)
|
2018-01-25 14:42:39 +00:00
|
|
|
{
|
2018-03-13 14:18:11 +00:00
|
|
|
setColumns(ColumnsDescription({
|
2017-04-17 15:06:12 +00:00
|
|
|
{ "database", std::make_shared<DataTypeString>() },
|
|
|
|
{ "table", std::make_shared<DataTypeString>() },
|
|
|
|
{ "engine", std::make_shared<DataTypeString>() },
|
|
|
|
{ "is_leader", std::make_shared<DataTypeUInt8>() },
|
2019-02-13 13:05:58 +00:00
|
|
|
{ "can_become_leader", std::make_shared<DataTypeUInt8>() },
|
2017-04-17 15:06:12 +00:00
|
|
|
{ "is_readonly", std::make_shared<DataTypeUInt8>() },
|
|
|
|
{ "is_session_expired", std::make_shared<DataTypeUInt8>() },
|
|
|
|
{ "future_parts", std::make_shared<DataTypeUInt32>() },
|
|
|
|
{ "parts_to_check", std::make_shared<DataTypeUInt32>() },
|
|
|
|
{ "zookeeper_path", std::make_shared<DataTypeString>() },
|
|
|
|
{ "replica_name", std::make_shared<DataTypeString>() },
|
|
|
|
{ "replica_path", std::make_shared<DataTypeString>() },
|
|
|
|
{ "columns_version", std::make_shared<DataTypeInt32>() },
|
|
|
|
{ "queue_size", std::make_shared<DataTypeUInt32>() },
|
|
|
|
{ "inserts_in_queue", std::make_shared<DataTypeUInt32>() },
|
|
|
|
{ "merges_in_queue", std::make_shared<DataTypeUInt32>() },
|
2018-06-06 13:22:30 +00:00
|
|
|
{ "part_mutations_in_queue", std::make_shared<DataTypeUInt32>() },
|
2017-04-17 15:06:12 +00:00
|
|
|
{ "queue_oldest_time", std::make_shared<DataTypeDateTime>() },
|
|
|
|
{ "inserts_oldest_time", std::make_shared<DataTypeDateTime>() },
|
|
|
|
{ "merges_oldest_time", std::make_shared<DataTypeDateTime>() },
|
2018-06-06 13:22:30 +00:00
|
|
|
{ "part_mutations_oldest_time", std::make_shared<DataTypeDateTime>() },
|
2017-04-17 15:06:12 +00:00
|
|
|
{ "oldest_part_to_get", std::make_shared<DataTypeString>() },
|
|
|
|
{ "oldest_part_to_merge_to", std::make_shared<DataTypeString>() },
|
2018-06-06 13:22:30 +00:00
|
|
|
{ "oldest_part_to_mutate_to", std::make_shared<DataTypeString>() },
|
2017-04-17 15:06:12 +00:00
|
|
|
{ "log_max_index", std::make_shared<DataTypeUInt64>() },
|
|
|
|
{ "log_pointer", std::make_shared<DataTypeUInt64>() },
|
|
|
|
{ "last_queue_update", std::make_shared<DataTypeDateTime>() },
|
|
|
|
{ "absolute_delay", std::make_shared<DataTypeUInt64>() },
|
|
|
|
{ "total_replicas", std::make_shared<DataTypeUInt8>() },
|
|
|
|
{ "active_replicas", std::make_shared<DataTypeUInt8>() },
|
2018-03-13 14:18:11 +00:00
|
|
|
}));
|
2014-10-06 05:18:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
BlockInputStreams StorageSystemReplicas::read(
|
2017-04-01 07:20:54 +00:00
|
|
|
const Names & column_names,
|
2017-07-15 03:48:36 +00:00
|
|
|
const SelectQueryInfo & query_info,
|
2017-04-01 07:20:54 +00:00
|
|
|
const Context & context,
|
2018-09-08 11:29:23 +00:00
|
|
|
QueryProcessingStage::Enum /*processed_stage*/,
|
2017-12-02 02:47:12 +00:00
|
|
|
const size_t /*max_block_size*/,
|
|
|
|
const unsigned /*num_streams*/)
|
2014-10-06 05:18:17 +00:00
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
check(column_names);
|
|
|
|
|
|
|
|
/// We collect a set of replicated tables.
|
|
|
|
std::map<String, std::map<String, StoragePtr>> replicated_tables;
|
|
|
|
for (const auto & db : context.getDatabases())
|
2018-08-13 09:11:58 +00:00
|
|
|
{
|
2019-10-01 12:44:17 +00:00
|
|
|
/// Lazy database can not contain replicated tables
|
|
|
|
if (db.second->getEngineName() == "Lazy")
|
|
|
|
continue;
|
2018-08-14 02:18:57 +00:00
|
|
|
if (context.hasDatabaseAccessRights(db.first))
|
2018-08-13 09:11:58 +00:00
|
|
|
{
|
|
|
|
for (auto iterator = db.second->getIterator(context); iterator->isValid(); iterator->next())
|
|
|
|
if (dynamic_cast<const StorageReplicatedMergeTree *>(iterator->table().get()))
|
|
|
|
replicated_tables[db.first][iterator->name()] = iterator->table();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2018-08-27 16:31:45 +00:00
|
|
|
/// Do you need columns that require a ZooKeeper request to compute.
|
2017-04-01 07:20:54 +00:00
|
|
|
bool with_zk_fields = false;
|
2018-08-27 16:31:45 +00:00
|
|
|
for (const auto & column_name : column_names)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2018-08-27 16:31:45 +00:00
|
|
|
if ( column_name == "log_max_index"
|
|
|
|
|| column_name == "log_pointer"
|
|
|
|
|| column_name == "total_replicas"
|
|
|
|
|| column_name == "active_replicas")
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
with_zk_fields = true;
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-12-16 00:49:03 +00:00
|
|
|
MutableColumnPtr col_database_mut = ColumnString::create();
|
|
|
|
MutableColumnPtr col_table_mut = ColumnString::create();
|
|
|
|
MutableColumnPtr col_engine_mut = ColumnString::create();
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
for (auto & db : replicated_tables)
|
|
|
|
{
|
|
|
|
for (auto & table : db.second)
|
|
|
|
{
|
2017-12-16 00:49:03 +00:00
|
|
|
col_database_mut->insert(db.first);
|
|
|
|
col_table_mut->insert(table.first);
|
|
|
|
col_engine_mut->insert(table.second->getName());
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-12-16 00:49:03 +00:00
|
|
|
ColumnPtr col_database = std::move(col_database_mut);
|
|
|
|
ColumnPtr col_table = std::move(col_table_mut);
|
|
|
|
ColumnPtr col_engine = std::move(col_engine_mut);
|
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
/// Determine what tables are needed by the conditions in the query.
|
|
|
|
{
|
2017-12-16 00:49:03 +00:00
|
|
|
Block filtered_block
|
|
|
|
{
|
|
|
|
{ col_database, std::make_shared<DataTypeString>(), "database" },
|
|
|
|
{ col_table, std::make_shared<DataTypeString>(), "table" },
|
|
|
|
{ col_engine, std::make_shared<DataTypeString>(), "engine" },
|
|
|
|
};
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2017-07-15 03:48:36 +00:00
|
|
|
VirtualColumnUtils::filterBlockWithQuery(query_info.query, filtered_block, context);
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
if (!filtered_block.rows())
|
|
|
|
return BlockInputStreams();
|
|
|
|
|
2017-12-16 00:49:03 +00:00
|
|
|
col_database = filtered_block.getByName("database").column;
|
|
|
|
col_table = filtered_block.getByName("table").column;
|
|
|
|
col_engine = filtered_block.getByName("engine").column;
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
2017-12-16 00:49:03 +00:00
|
|
|
MutableColumns res_columns = getSampleBlock().cloneEmptyColumns();
|
|
|
|
|
|
|
|
for (size_t i = 0, size = col_database->size(); i < size; ++i)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
StorageReplicatedMergeTree::Status status;
|
2017-11-04 16:46:14 +00:00
|
|
|
dynamic_cast<StorageReplicatedMergeTree &>(
|
2017-04-01 07:20:54 +00:00
|
|
|
*replicated_tables
|
2017-12-16 00:49:03 +00:00
|
|
|
[(*col_database)[i].safeGet<const String &>()]
|
|
|
|
[(*col_table)[i].safeGet<const String &>()]).getStatus(status, with_zk_fields);
|
|
|
|
|
|
|
|
size_t col_num = 3;
|
2018-10-22 08:54:54 +00:00
|
|
|
res_columns[col_num++]->insert(status.is_leader);
|
2019-02-13 13:05:58 +00:00
|
|
|
res_columns[col_num++]->insert(status.can_become_leader);
|
2018-10-22 08:54:54 +00:00
|
|
|
res_columns[col_num++]->insert(status.is_readonly);
|
|
|
|
res_columns[col_num++]->insert(status.is_session_expired);
|
|
|
|
res_columns[col_num++]->insert(status.queue.future_parts);
|
|
|
|
res_columns[col_num++]->insert(status.parts_to_check);
|
2017-12-16 00:49:03 +00:00
|
|
|
res_columns[col_num++]->insert(status.zookeeper_path);
|
|
|
|
res_columns[col_num++]->insert(status.replica_name);
|
|
|
|
res_columns[col_num++]->insert(status.replica_path);
|
2018-10-22 08:54:54 +00:00
|
|
|
res_columns[col_num++]->insert(status.columns_version);
|
|
|
|
res_columns[col_num++]->insert(status.queue.queue_size);
|
|
|
|
res_columns[col_num++]->insert(status.queue.inserts_in_queue);
|
|
|
|
res_columns[col_num++]->insert(status.queue.merges_in_queue);
|
|
|
|
res_columns[col_num++]->insert(status.queue.part_mutations_in_queue);
|
|
|
|
res_columns[col_num++]->insert(status.queue.queue_oldest_time);
|
|
|
|
res_columns[col_num++]->insert(status.queue.inserts_oldest_time);
|
|
|
|
res_columns[col_num++]->insert(status.queue.merges_oldest_time);
|
|
|
|
res_columns[col_num++]->insert(status.queue.part_mutations_oldest_time);
|
2017-12-16 00:49:03 +00:00
|
|
|
res_columns[col_num++]->insert(status.queue.oldest_part_to_get);
|
|
|
|
res_columns[col_num++]->insert(status.queue.oldest_part_to_merge_to);
|
2018-06-06 13:22:30 +00:00
|
|
|
res_columns[col_num++]->insert(status.queue.oldest_part_to_mutate_to);
|
2017-12-16 00:49:03 +00:00
|
|
|
res_columns[col_num++]->insert(status.log_max_index);
|
|
|
|
res_columns[col_num++]->insert(status.log_pointer);
|
2018-10-22 08:54:54 +00:00
|
|
|
res_columns[col_num++]->insert(status.queue.last_queue_update);
|
|
|
|
res_columns[col_num++]->insert(status.absolute_delay);
|
|
|
|
res_columns[col_num++]->insert(status.total_replicas);
|
|
|
|
res_columns[col_num++]->insert(status.active_replicas);
|
2017-12-16 00:49:03 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
Block res = getSampleBlock().cloneEmpty();
|
|
|
|
size_t col_num = 0;
|
|
|
|
res.getByPosition(col_num++).column = col_database;
|
|
|
|
res.getByPosition(col_num++).column = col_table;
|
|
|
|
res.getByPosition(col_num++).column = col_engine;
|
|
|
|
size_t num_columns = res.columns();
|
|
|
|
while (col_num < num_columns)
|
|
|
|
{
|
|
|
|
res.getByPosition(col_num).column = std::move(res_columns[col_num]);
|
|
|
|
++col_num;
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
2017-12-16 00:49:03 +00:00
|
|
|
return BlockInputStreams(1, std::make_shared<OneBlockInputStream>(res));
|
2014-10-06 05:18:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
}
|