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
|
#include <Storages/System/StorageSystemAsynchronousInserts.h>
#include <Processors/Sources/SourceFromSingleChunk.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/DataTypeArray.h>
#include <DataTypes/DataTypeDateTime64.h>
#include <DataTypes/DataTypesNumber.h>
#include <Interpreters/AsynchronousInsertQueue.h>
#include <Interpreters/Context.h>
#include <Parsers/queryToString.h>
#include <Core/DecimalFunctions.h>
#include <Parsers/ASTInsertQuery.h>
namespace DB
{
static constexpr auto TIME_SCALE = 6;
NamesAndTypesList StorageSystemAsynchronousInserts::getNamesAndTypes()
{
return
{
{"query", std::make_shared<DataTypeString>()},
{"database", std::make_shared<DataTypeString>()},
{"table", std::make_shared<DataTypeString>()},
{"format", std::make_shared<DataTypeString>()},
{"first_update", std::make_shared<DataTypeDateTime64>(TIME_SCALE)},
{"total_bytes", std::make_shared<DataTypeUInt64>()},
{"entries.query_id", std::make_shared<DataTypeArray>(std::make_shared<DataTypeString>())},
{"entries.bytes", std::make_shared<DataTypeArray>(std::make_shared<DataTypeUInt64>())},
};
}
void StorageSystemAsynchronousInserts::fillData(MutableColumns & res_columns, ContextPtr context, const SelectQueryInfo &) const
{
using namespace std::chrono;
auto * insert_queue = context->getAsynchronousInsertQueue();
if (!insert_queue)
return;
for (size_t shard_num = 0; shard_num < insert_queue->getPoolSize(); ++shard_num)
{
auto [queue, queue_lock] = insert_queue->getQueueLocked(shard_num);
for (const auto & [first_update, elem] : queue)
{
const auto & [key, data] = elem;
auto time_in_microseconds = [](const time_point<steady_clock> & timestamp)
{
auto time_diff = duration_cast<microseconds>(steady_clock::now() - timestamp);
auto time_us = (system_clock::now() - time_diff).time_since_epoch().count();
DecimalUtils::DecimalComponents<DateTime64> components{time_us / 1'000'000, time_us % 1'000'000};
return DecimalField(DecimalUtils::decimalFromComponents<DateTime64>(components, TIME_SCALE), TIME_SCALE);
};
const auto & insert_query = key.query->as<const ASTInsertQuery &>();
size_t i = 0;
res_columns[i++]->insert(queryToString(insert_query));
/// If query is "INSERT INTO FUNCTION" then table_id is empty.
if (insert_query.table_id)
{
res_columns[i++]->insert(insert_query.table_id.getDatabaseName());
res_columns[i++]->insert(insert_query.table_id.getTableName());
}
else
{
res_columns[i++]->insertDefault();
res_columns[i++]->insertDefault();
}
res_columns[i++]->insert(insert_query.format);
res_columns[i++]->insert(time_in_microseconds(first_update));
res_columns[i++]->insert(data->size_in_bytes);
Array arr_query_id;
Array arr_bytes;
for (const auto & entry : data->entries)
{
arr_query_id.push_back(entry->query_id);
arr_bytes.push_back(entry->bytes.size());
}
res_columns[i++]->insert(arr_query_id);
res_columns[i++]->insert(arr_bytes);
}
}
}
}
|
