#include #include #include #include #include #include #include #include #include #include #include #include #include namespace DB { NamesAndTypesList StorageSystemReplicationQueue::getNamesAndTypes() { return { /// Table properties. { "database", std::make_shared() }, { "table", std::make_shared() }, { "replica_name", std::make_shared() }, /// Constant element properties. { "position", std::make_shared() }, { "node_name", std::make_shared() }, { "type", std::make_shared() }, { "create_time", std::make_shared() }, { "required_quorum", std::make_shared() }, { "source_replica", std::make_shared() }, { "new_part_name", std::make_shared() }, { "parts_to_merge", std::make_shared(std::make_shared()) }, { "is_detach", std::make_shared() }, /// Processing status of item. { "is_currently_executing", std::make_shared() }, { "num_tries", std::make_shared() }, { "last_exception", std::make_shared() }, { "last_attempt_time", std::make_shared() }, { "num_postponed", std::make_shared() }, { "postpone_reason", std::make_shared() }, { "last_postpone_time", std::make_shared() }, { "merge_type", std::make_shared() }, }; } void StorageSystemReplicationQueue::fillData(MutableColumns & res_columns, const Context & context, const SelectQueryInfo & query_info) const { const auto access = context.getAccess(); const bool check_access_for_databases = !access->isGranted(AccessType::SHOW_TABLES); std::map> replicated_tables; for (const auto & db : DatabaseCatalog::instance().getDatabases()) { /// Check if database can contain replicated tables if (!db.second->canContainMergeTreeTables()) continue; const bool check_access_for_tables = check_access_for_databases && !access->isGranted(AccessType::SHOW_TABLES, db.first); for (auto iterator = db.second->getTablesIterator(context); iterator->isValid(); iterator->next()) { const auto & table = iterator->table(); if (!table) continue; if (!dynamic_cast(table.get())) continue; if (check_access_for_tables && !access->isGranted(AccessType::SHOW_TABLES, db.first, iterator->name())) continue; replicated_tables[db.first][iterator->name()] = table; } } MutableColumnPtr col_database_mut = ColumnString::create(); MutableColumnPtr col_table_mut = ColumnString::create(); for (auto & db : replicated_tables) { for (auto & table : db.second) { col_database_mut->insert(db.first); col_table_mut->insert(table.first); } } ColumnPtr col_database_to_filter = std::move(col_database_mut); ColumnPtr col_table_to_filter = std::move(col_table_mut); /// Determine what tables are needed by the conditions in the query. { Block filtered_block { { col_database_to_filter, std::make_shared(), "database" }, { col_table_to_filter, std::make_shared(), "table" }, }; VirtualColumnUtils::filterBlockWithQuery(query_info.query, filtered_block, context); if (!filtered_block.rows()) return; col_database_to_filter = filtered_block.getByName("database").column; col_table_to_filter = filtered_block.getByName("table").column; } StorageReplicatedMergeTree::LogEntriesData queue; String replica_name; for (size_t i = 0, tables_size = col_database_to_filter->size(); i < tables_size; ++i) { String database = (*col_database_to_filter)[i].safeGet(); String table = (*col_table_to_filter)[i].safeGet(); dynamic_cast(*replicated_tables[database][table]).getQueue(queue, replica_name); for (size_t j = 0, queue_size = queue.size(); j < queue_size; ++j) { const auto & entry = queue[j]; Array parts_to_merge; parts_to_merge.reserve(entry.source_parts.size()); for (const auto & part_name : entry.source_parts) parts_to_merge.push_back(part_name); size_t col_num = 0; res_columns[col_num++]->insert(database); res_columns[col_num++]->insert(table); res_columns[col_num++]->insert(replica_name); res_columns[col_num++]->insert(j); res_columns[col_num++]->insert(entry.znode_name); res_columns[col_num++]->insert(entry.typeToString()); res_columns[col_num++]->insert(entry.create_time); res_columns[col_num++]->insert(entry.quorum); res_columns[col_num++]->insert(entry.source_replica); res_columns[col_num++]->insert(entry.new_part_name); res_columns[col_num++]->insert(parts_to_merge); res_columns[col_num++]->insert(entry.detach); res_columns[col_num++]->insert(entry.currently_executing); res_columns[col_num++]->insert(entry.num_tries); res_columns[col_num++]->insert(entry.exception ? getExceptionMessage(entry.exception, false) : ""); res_columns[col_num++]->insert(UInt64(entry.last_attempt_time)); res_columns[col_num++]->insert(entry.num_postponed); res_columns[col_num++]->insert(entry.postpone_reason); res_columns[col_num++]->insert(UInt64(entry.last_postpone_time)); if (entry.type == ReplicatedMergeTreeLogEntryData::Type::MERGE_PARTS) res_columns[col_num++]->insert(toString(entry.merge_type)); else res_columns[col_num++]->insertDefault(); } } } }