1#include <iomanip>
2
3#include <Core/Settings.h>
4#include <Databases/DatabaseOnDisk.h>
5#include <Databases/DatabaseOrdinary.h>
6#include <Databases/DatabasesCommon.h>
7#include <IO/ReadBufferFromFile.h>
8#include <IO/ReadHelpers.h>
9#include <IO/WriteBufferFromFile.h>
10#include <IO/WriteHelpers.h>
11#include <Interpreters/Context.h>
12#include <Interpreters/InterpreterCreateQuery.h>
13#include <Parsers/ASTCreateQuery.h>
14#include <Parsers/ParserCreateQuery.h>
15#include <Storages/StorageFactory.h>
16#include <Parsers/parseQuery.h>
17#include <Parsers/formatAST.h>
18#include <Parsers/ASTSetQuery.h>
19#include <TableFunctions/TableFunctionFactory.h>
20
21#include <Parsers/queryToString.h>
22
23#include <Poco/DirectoryIterator.h>
24#include <Poco/Event.h>
25#include <Common/Stopwatch.h>
26#include <Common/quoteString.h>
27#include <Common/ThreadPool.h>
28#include <Common/escapeForFileName.h>
29#include <Common/typeid_cast.h>
30#include <common/logger_useful.h>
31#include <ext/scope_guard.h>
32
33
34namespace DB
35{
36
37namespace ErrorCodes
38{
39 extern const int CANNOT_CREATE_TABLE_FROM_METADATA;
40 extern const int CANNOT_CREATE_DICTIONARY_FROM_METADATA;
41 extern const int EMPTY_LIST_OF_COLUMNS_PASSED;
42 extern const int CANNOT_PARSE_TEXT;
43}
44
45
46static constexpr size_t PRINT_MESSAGE_EACH_N_OBJECTS = 256;
47static constexpr size_t PRINT_MESSAGE_EACH_N_SECONDS = 5;
48static constexpr size_t METADATA_FILE_BUFFER_SIZE = 32768;
49
50
51namespace
52{
53 void tryAttachTable(
54 Context & context,
55 const ASTCreateQuery & query,
56 DatabaseOrdinary & database,
57 const String & database_name,
58 bool has_force_restore_data_flag)
59 {
60 assert(!query.is_dictionary);
61 try
62 {
63 String table_name;
64 StoragePtr table;
65 std::tie(table_name, table)
66 = createTableFromAST(query, database_name, database.getTableDataPath(query), context, has_force_restore_data_flag);
67 database.attachTable(table_name, table);
68 }
69 catch (const Exception & e)
70 {
71 throw Exception(
72 "Cannot attach table '" + query.table + "' from query " + serializeAST(query)
73 + ". Error: " + DB::getCurrentExceptionMessage(true),
74 e,
75 DB::ErrorCodes::CANNOT_CREATE_TABLE_FROM_METADATA);
76 }
77 }
78
79
80 void tryAttachDictionary(
81 Context & context,
82 const ASTCreateQuery & query,
83 DatabaseOrdinary & database)
84 {
85 assert(query.is_dictionary);
86 try
87 {
88 database.attachDictionary(query.table, context);
89 }
90 catch (const Exception & e)
91 {
92 throw Exception(
93 "Cannot create dictionary '" + query.table + "' from query " + serializeAST(query)
94 + ". Error: " + DB::getCurrentExceptionMessage(true),
95 e,
96 DB::ErrorCodes::CANNOT_CREATE_DICTIONARY_FROM_METADATA);
97 }
98 }
99
100
101 void logAboutProgress(Poco::Logger * log, size_t processed, size_t total, AtomicStopwatch & watch)
102 {
103 if (processed % PRINT_MESSAGE_EACH_N_OBJECTS == 0 || watch.compareAndRestart(PRINT_MESSAGE_EACH_N_SECONDS))
104 {
105 LOG_INFO(log, std::fixed << std::setprecision(2) << processed * 100.0 / total << "%");
106 watch.restart();
107 }
108 }
109}
110
111
112DatabaseOrdinary::DatabaseOrdinary(const String & name_, const String & metadata_path_, const Context & context_)
113 : DatabaseWithDictionaries(name_, metadata_path_, "DatabaseOrdinary (" + name_ + ")")
114{
115 Poco::File(context_.getPath() + getDataPath()).createDirectories();
116}
117
118
119void DatabaseOrdinary::loadStoredObjects(
120 Context & context,
121 bool has_force_restore_data_flag)
122{
123
124 /** Tables load faster if they are loaded in sorted (by name) order.
125 * Otherwise (for the ext4 filesystem), `DirectoryIterator` iterates through them in some order,
126 * which does not correspond to order tables creation and does not correspond to order of their location on disk.
127 */
128 using FileNames = std::map<std::string, ASTPtr>;
129 FileNames file_names;
130
131 size_t total_dictionaries = 0;
132 iterateMetadataFiles(context, [&file_names, &total_dictionaries, this](const String & file_name)
133 {
134 String full_path = getMetadataPath() + file_name;
135 try
136 {
137 auto ast = parseQueryFromMetadata(full_path, /*throw_on_error*/ true, /*remove_empty*/false);
138 if (ast)
139 {
140 auto * create_query = ast->as<ASTCreateQuery>();
141 file_names[file_name] = ast;
142 total_dictionaries += create_query->is_dictionary;
143 }
144 }
145 catch (const Exception & e)
146 {
147 throw Exception(
148 "Cannot parse definition from metadata file " + full_path + ". Error: " + DB::getCurrentExceptionMessage(true), e, ErrorCodes::CANNOT_PARSE_TEXT);
149 }
150
151 });
152
153 size_t total_tables = file_names.size() - total_dictionaries;
154
155 LOG_INFO(log, "Total " << total_tables << " tables and " << total_dictionaries << " dictionaries.");
156
157 AtomicStopwatch watch;
158 std::atomic<size_t> tables_processed{0};
159 std::atomic<size_t> dictionaries_processed{0};
160
161 ThreadPool pool(SettingMaxThreads().getAutoValue());
162
163 /// Attach tables.
164 for (const auto & name_with_query : file_names)
165 {
166 const auto & create_query = name_with_query.second->as<const ASTCreateQuery &>();
167 if (!create_query.is_dictionary)
168 pool.scheduleOrThrowOnError([&]()
169 {
170 tryAttachTable(context, create_query, *this, getDatabaseName(), has_force_restore_data_flag);
171
172 /// Messages, so that it's not boring to wait for the server to load for a long time.
173 logAboutProgress(log, ++tables_processed, total_tables, watch);
174 });
175 }
176
177 pool.wait();
178
179 /// After all tables was basically initialized, startup them.
180 startupTables(pool);
181
182 /// Attach dictionaries.
183 attachToExternalDictionariesLoader(context);
184 for (const auto & name_with_query : file_names)
185 {
186 auto create_query = name_with_query.second->as<const ASTCreateQuery &>();
187 if (create_query.is_dictionary)
188 {
189 tryAttachDictionary(context, create_query, *this);
190
191 /// Messages, so that it's not boring to wait for the server to load for a long time.
192 logAboutProgress(log, ++dictionaries_processed, total_dictionaries, watch);
193 }
194 }
195}
196
197
198void DatabaseOrdinary::startupTables(ThreadPool & thread_pool)
199{
200 LOG_INFO(log, "Starting up tables.");
201
202 const size_t total_tables = tables.size();
203 if (!total_tables)
204 return;
205
206 AtomicStopwatch watch;
207 std::atomic<size_t> tables_processed{0};
208
209 auto startupOneTable = [&](const StoragePtr & table)
210 {
211 table->startup();
212 logAboutProgress(log, ++tables_processed, total_tables, watch);
213 };
214
215 try
216 {
217 for (const auto & table : tables)
218 thread_pool.scheduleOrThrowOnError([&]() { startupOneTable(table.second); });
219 }
220 catch (...)
221 {
222 thread_pool.wait();
223 throw;
224 }
225 thread_pool.wait();
226}
227
228void DatabaseOrdinary::alterTable(
229 const Context & context,
230 const String & table_name,
231 const StorageInMemoryMetadata & metadata)
232{
233 /// Read the definition of the table and replace the necessary parts with new ones.
234 String table_metadata_path = getObjectMetadataPath(table_name);
235 String table_metadata_tmp_path = table_metadata_path + ".tmp";
236 String statement;
237
238 {
239 char in_buf[METADATA_FILE_BUFFER_SIZE];
240 ReadBufferFromFile in(table_metadata_path, METADATA_FILE_BUFFER_SIZE, -1, in_buf);
241 readStringUntilEOF(statement, in);
242 }
243
244 ParserCreateQuery parser;
245 ASTPtr ast = parseQuery(parser, statement.data(), statement.data() + statement.size(), "in file " + table_metadata_path, 0);
246
247 const auto & ast_create_query = ast->as<ASTCreateQuery &>();
248
249 ASTPtr new_columns = InterpreterCreateQuery::formatColumns(metadata.columns);
250 ASTPtr new_indices = InterpreterCreateQuery::formatIndices(metadata.indices);
251 ASTPtr new_constraints = InterpreterCreateQuery::formatConstraints(metadata.constraints);
252
253 ast_create_query.columns_list->replace(ast_create_query.columns_list->columns, new_columns);
254 ast_create_query.columns_list->setOrReplace(ast_create_query.columns_list->indices, new_indices);
255 ast_create_query.columns_list->setOrReplace(ast_create_query.columns_list->constraints, new_constraints);
256
257 ASTStorage & storage_ast = *ast_create_query.storage;
258 /// ORDER BY may change, but cannot appear, it's required construction
259 if (metadata.order_by_ast && storage_ast.order_by)
260 storage_ast.set(storage_ast.order_by, metadata.order_by_ast);
261
262 if (metadata.primary_key_ast)
263 storage_ast.set(storage_ast.primary_key, metadata.primary_key_ast);
264
265 if (metadata.ttl_for_table_ast)
266 storage_ast.set(storage_ast.ttl_table, metadata.ttl_for_table_ast);
267
268 if (metadata.settings_ast)
269 storage_ast.set(storage_ast.settings, metadata.settings_ast);
270
271
272 statement = getObjectDefinitionFromCreateQuery(ast);
273 {
274 WriteBufferFromFile out(table_metadata_tmp_path, statement.size(), O_WRONLY | O_CREAT | O_EXCL);
275 writeString(statement, out);
276 out.next();
277 if (context.getSettingsRef().fsync_metadata)
278 out.sync();
279 out.close();
280 }
281
282 try
283 {
284 /// rename atomically replaces the old file with the new one.
285 Poco::File(table_metadata_tmp_path).renameTo(table_metadata_path);
286 }
287 catch (...)
288 {
289 Poco::File(table_metadata_tmp_path).remove();
290 throw;
291 }
292}
293
294}
295