1#include <Columns/ColumnString.h>
2#include <Columns/ColumnArray.h>
3#include <Columns/ColumnsNumber.h>
4#include <DataTypes/DataTypeString.h>
5#include <DataTypes/DataTypesNumber.h>
6#include <DataTypes/DataTypeDateTime.h>
7#include <DataTypes/DataTypeArray.h>
8#include <Storages/System/StorageSystemReplicationQueue.h>
9#include <Storages/StorageReplicatedMergeTree.h>
10#include <Storages/VirtualColumnUtils.h>
11#include <Common/typeid_cast.h>
12#include <Databases/IDatabase.h>
13
14
15namespace DB
16{
17
18
19
20NamesAndTypesList StorageSystemReplicationQueue::getNamesAndTypes()
21{
22 return {
23 /// Table properties.
24 { "database", std::make_shared<DataTypeString>() },
25 { "table", std::make_shared<DataTypeString>() },
26 { "replica_name", std::make_shared<DataTypeString>() },
27 /// Constant element properties.
28 { "position", std::make_shared<DataTypeUInt32>() },
29 { "node_name", std::make_shared<DataTypeString>() },
30 { "type", std::make_shared<DataTypeString>() },
31 { "create_time", std::make_shared<DataTypeDateTime>() },
32 { "required_quorum", std::make_shared<DataTypeUInt32>() },
33 { "source_replica", std::make_shared<DataTypeString>() },
34 { "new_part_name", std::make_shared<DataTypeString>() },
35 { "parts_to_merge", std::make_shared<DataTypeArray>(std::make_shared<DataTypeString>()) },
36 { "is_detach", std::make_shared<DataTypeUInt8>() },
37 /// Processing status of item.
38 { "is_currently_executing", std::make_shared<DataTypeUInt8>() },
39 { "num_tries", std::make_shared<DataTypeUInt32>() },
40 { "last_exception", std::make_shared<DataTypeString>() },
41 { "last_attempt_time", std::make_shared<DataTypeDateTime>() },
42 { "num_postponed", std::make_shared<DataTypeUInt32>() },
43 { "postpone_reason", std::make_shared<DataTypeString>() },
44 { "last_postpone_time", std::make_shared<DataTypeDateTime>() },
45 };
46}
47
48
49void StorageSystemReplicationQueue::fillData(MutableColumns & res_columns, const Context & context, const SelectQueryInfo & query_info) const
50{
51 std::map<String, std::map<String, StoragePtr>> replicated_tables;
52 for (const auto & db : context.getDatabases())
53 {
54 /// Lazy database can not contain replicated tables
55 if (db.second->getEngineName() == "Lazy")
56 continue;
57
58 if (context.hasDatabaseAccessRights(db.first))
59 {
60 for (auto iterator = db.second->getTablesIterator(context); iterator->isValid(); iterator->next())
61 if (dynamic_cast<const StorageReplicatedMergeTree *>(iterator->table().get()))
62 replicated_tables[db.first][iterator->name()] = iterator->table();
63 }
64 }
65
66
67 MutableColumnPtr col_database_mut = ColumnString::create();
68 MutableColumnPtr col_table_mut = ColumnString::create();
69
70 for (auto & db : replicated_tables)
71 {
72 for (auto & table : db.second)
73 {
74 col_database_mut->insert(db.first);
75 col_table_mut->insert(table.first);
76 }
77 }
78
79 ColumnPtr col_database_to_filter = std::move(col_database_mut);
80 ColumnPtr col_table_to_filter = std::move(col_table_mut);
81
82 /// Determine what tables are needed by the conditions in the query.
83 {
84 Block filtered_block
85 {
86 { col_database_to_filter, std::make_shared<DataTypeString>(), "database" },
87 { col_table_to_filter, std::make_shared<DataTypeString>(), "table" },
88 };
89
90 VirtualColumnUtils::filterBlockWithQuery(query_info.query, filtered_block, context);
91
92 if (!filtered_block.rows())
93 return;
94
95 col_database_to_filter = filtered_block.getByName("database").column;
96 col_table_to_filter = filtered_block.getByName("table").column;
97 }
98
99 StorageReplicatedMergeTree::LogEntriesData queue;
100 String replica_name;
101
102 for (size_t i = 0, tables_size = col_database_to_filter->size(); i < tables_size; ++i)
103 {
104 String database = (*col_database_to_filter)[i].safeGet<const String &>();
105 String table = (*col_table_to_filter)[i].safeGet<const String &>();
106
107 dynamic_cast<StorageReplicatedMergeTree &>(*replicated_tables[database][table]).getQueue(queue, replica_name);
108
109 for (size_t j = 0, queue_size = queue.size(); j < queue_size; ++j)
110 {
111 const auto & entry = queue[j];
112
113 Array parts_to_merge;
114 parts_to_merge.reserve(entry.source_parts.size());
115 for (const auto & part_name : entry.source_parts)
116 parts_to_merge.push_back(part_name);
117
118 size_t col_num = 0;
119 res_columns[col_num++]->insert(database);
120 res_columns[col_num++]->insert(table);
121 res_columns[col_num++]->insert(replica_name);
122 res_columns[col_num++]->insert(j);
123 res_columns[col_num++]->insert(entry.znode_name);
124 res_columns[col_num++]->insert(entry.typeToString());
125 res_columns[col_num++]->insert(entry.create_time);
126 res_columns[col_num++]->insert(entry.quorum);
127 res_columns[col_num++]->insert(entry.source_replica);
128 res_columns[col_num++]->insert(entry.new_part_name);
129 res_columns[col_num++]->insert(parts_to_merge);
130 res_columns[col_num++]->insert(entry.detach);
131 res_columns[col_num++]->insert(entry.currently_executing);
132 res_columns[col_num++]->insert(entry.num_tries);
133 res_columns[col_num++]->insert(entry.exception ? getExceptionMessage(entry.exception, false) : "");
134 res_columns[col_num++]->insert(UInt64(entry.last_attempt_time));
135 res_columns[col_num++]->insert(entry.num_postponed);
136 res_columns[col_num++]->insert(entry.postpone_reason);
137 res_columns[col_num++]->insert(UInt64(entry.last_postpone_time));
138 }
139 }
140}
141
142}
143