1 | #include <Interpreters/Context.h> |
2 | #include <Interpreters/InterpreterCheckQuery.h> |
3 | #include <Storages/IStorage.h> |
4 | #include <Parsers/ASTCheckQuery.h> |
5 | #include <DataStreams/OneBlockInputStream.h> |
6 | #include <DataTypes/DataTypesNumber.h> |
7 | #include <DataTypes/DataTypeString.h> |
8 | #include <Columns/ColumnsNumber.h> |
9 | #include <Common/typeid_cast.h> |
10 | #include <algorithm> |
11 | |
12 | |
13 | namespace DB |
14 | { |
15 | |
16 | namespace |
17 | { |
18 | |
19 | NamesAndTypes getBlockStructure() |
20 | { |
21 | return { |
22 | {"part_path" , std::make_shared<DataTypeString>()}, |
23 | {"is_passed" , std::make_shared<DataTypeUInt8>()}, |
24 | {"message" , std::make_shared<DataTypeString>()}, |
25 | }; |
26 | } |
27 | |
28 | } |
29 | |
30 | |
31 | InterpreterCheckQuery::InterpreterCheckQuery(const ASTPtr & query_ptr_, const Context & context_) |
32 | : query_ptr(query_ptr_), context(context_) |
33 | { |
34 | } |
35 | |
36 | |
37 | BlockIO InterpreterCheckQuery::execute() |
38 | { |
39 | const auto & check = query_ptr->as<ASTCheckQuery &>(); |
40 | const String & table_name = check.table; |
41 | String database_name = check.database.empty() ? context.getCurrentDatabase() : check.database; |
42 | |
43 | StoragePtr table = context.getTable(database_name, table_name); |
44 | auto check_results = table->checkData(query_ptr, context); |
45 | |
46 | Block block; |
47 | if (context.getSettingsRef().check_query_single_value_result) |
48 | { |
49 | bool result = std::all_of(check_results.begin(), check_results.end(), [] (const CheckResult & res) { return res.success; }); |
50 | auto column = ColumnUInt8::create(); |
51 | column->insertValue(UInt64(result)); |
52 | block = Block{{std::move(column), std::make_shared<DataTypeUInt8>(), "result" }}; |
53 | } |
54 | else |
55 | { |
56 | auto block_structure = getBlockStructure(); |
57 | auto path_column = block_structure[0].type->createColumn(); |
58 | auto is_passed_column = block_structure[1].type->createColumn(); |
59 | auto message_column = block_structure[2].type->createColumn(); |
60 | |
61 | for (const auto & check_result : check_results) |
62 | { |
63 | path_column->insert(check_result.fs_path); |
64 | is_passed_column->insert(static_cast<UInt8>(check_result.success)); |
65 | message_column->insert(check_result.failure_message); |
66 | } |
67 | |
68 | block = Block({ |
69 | {std::move(path_column), block_structure[0].type, block_structure[0].name}, |
70 | {std::move(is_passed_column), block_structure[1].type, block_structure[1].name}, |
71 | {std::move(message_column), block_structure[2].type, block_structure[2].name}}); |
72 | } |
73 | |
74 | BlockIO res; |
75 | res.in = std::make_shared<OneBlockInputStream>(block); |
76 | |
77 | return res; |
78 | } |
79 | |
80 | } |
81 | |