1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
#pragma once
#include "clickhouse_config.h"
#if USE_MYSQL
#error #include <mysqlxx/Pool.h>
#include <Core/MultiEnum.h>
#include <Core/NamesAndTypes.h>
#include <Common/ThreadPool.h>
#include <Storages/ColumnsDescription.h>
#include <Storages/MySQL/MySQLSettings.h>
#include <Databases/DatabasesCommon.h>
#include <Parsers/ASTCreateQuery.h>
#error #include <mysqlxx/PoolWithFailover.h>
#include <atomic>
#include <condition_variable>
#include <map>
#include <memory>
#include <mutex>
#include <unordered_set>
#include <vector>
namespace DB
{
class Context;
enum class MySQLDataTypesSupport;
/** Real-time access to table list and table structure from remote MySQL
* It doesn't make any manipulations with filesystem.
* All tables are created by calling code after real-time pull-out structure from remote MySQL
*/
class DatabaseMySQL final : public IDatabase, WithContext
{
public:
~DatabaseMySQL() override;
DatabaseMySQL(
ContextPtr context,
const String & database_name,
const String & metadata_path,
const ASTStorage * database_engine_define,
const String & database_name_in_mysql,
std::unique_ptr<MySQLSettings> settings_,
mysqlxx::PoolWithFailover && pool,
bool attach);
String getEngineName() const override { return "MySQL"; }
bool canContainMergeTreeTables() const override { return false; }
bool canContainDistributedTables() const override { return false; }
bool shouldBeEmptyOnDetach() const override { return false; }
bool empty() const override;
DatabaseTablesIteratorPtr getTablesIterator(ContextPtr context, const FilterByNameFunction & filter_by_table_name) const override;
ASTPtr getCreateDatabaseQuery() const override;
bool isTableExist(const String & name, ContextPtr context) const override;
StoragePtr tryGetTable(const String & name, ContextPtr context) const override;
time_t getObjectMetadataModificationTime(const String & name) const override;
void shutdown() override;
void drop(ContextPtr /*context*/) override;
String getMetadataPath() const override;
void createTable(ContextPtr, const String & table_name, const StoragePtr & storage, const ASTPtr & create_query) override;
void loadStoredObjects(ContextMutablePtr, LoadingStrictnessLevel /*mode*/) override;
StoragePtr detachTable(ContextPtr context, const String & table_name) override;
void detachTablePermanently(ContextPtr context, const String & table_name) override;
void dropTable(ContextPtr context, const String & table_name, bool sync) override;
void attachTable(ContextPtr context, const String & table_name, const StoragePtr & storage, const String & relative_table_path) override;
protected:
ASTPtr getCreateTableQueryImpl(const String & name, ContextPtr context, bool throw_on_error) const override;
private:
String metadata_path;
ASTPtr database_engine_define;
String database_name_in_mysql;
std::unique_ptr<MySQLSettings> mysql_settings;
std::atomic<bool> quit{false};
std::condition_variable cond;
using MySQLPool = mysqlxx::PoolWithFailover;
using ModifyTimeAndStorage = std::pair<UInt64, StoragePtr>;
mutable MySQLPool mysql_pool;
mutable std::vector<StoragePtr> outdated_tables;
mutable std::map<String, ModifyTimeAndStorage> local_tables_cache;
std::unordered_set<String> remove_or_detach_tables;
void cleanOutdatedTables();
void fetchTablesIntoLocalCache(ContextPtr context) const TSA_REQUIRES(mutex);
std::map<String, UInt64> fetchTablesWithModificationTime(ContextPtr local_context) const;
std::map<String, ColumnsDescription> fetchTablesColumnsList(const std::vector<String> & tables_name, ContextPtr context) const;
void destroyLocalCacheExtraTables(const std::map<String, UInt64> & tables_with_modification_time) const TSA_REQUIRES(mutex);
void fetchLatestTablesStructureIntoCache(const std::map<String, UInt64> & tables_modification_time, ContextPtr context) const TSA_REQUIRES(mutex);
ThreadFromGlobalPool thread;
};
}
#endif
|
