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 | |
15 | namespace DB |
16 | { |
17 | |
18 | |
19 | |
20 | NamesAndTypesList 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 | |
49 | void 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 | |