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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
#pragma once
#include <Storages/MergeTree/MergeTreeDataPartType.h>
#include <Interpreters/SystemLog.h>
#include <Core/NamesAndTypes.h>
#include <Core/NamesAndAliases.h>
#include <Core/UUID.h>
#include <Storages/MergeTree/MergeType.h>
#include <Storages/MergeTree/MergeAlgorithm.h>
namespace ProfileEvents
{
class Counters;
}
namespace DB
{
struct PartLogElement
{
enum Type
{
NEW_PART = 1,
MERGE_PARTS = 2,
DOWNLOAD_PART = 3,
REMOVE_PART = 4,
MUTATE_PART = 5,
MOVE_PART = 6,
};
/// Copy of MergeAlgorithm since values are written to disk.
enum PartMergeAlgorithm
{
UNDECIDED = 0,
VERTICAL = 1,
HORIZONTAL = 2,
};
enum MergeReasonType
{
/// merge_reason is relevant only for event_type = 'MERGE_PARTS', in other cases it is NOT_A_MERGE
NOT_A_MERGE = 1,
/// Just regular merge
REGULAR_MERGE = 2,
/// Merge assigned to delete some data from parts (with TTLMergeSelector)
TTL_DELETE_MERGE = 3,
/// Merge with recompression
TTL_RECOMPRESS_MERGE = 4,
};
String query_id;
Type event_type = NEW_PART;
MergeReasonType merge_reason = NOT_A_MERGE;
PartMergeAlgorithm merge_algorithm = UNDECIDED;
time_t event_time = 0;
Decimal64 event_time_microseconds = 0;
UInt64 duration_ms = 0;
String database_name;
String table_name;
UUID table_uuid{UUIDHelpers::Nil};
String part_name;
String partition_id;
String partition;
String disk_name;
String path_on_disk;
MergeTreeDataPartType part_type;
/// Size of the part
UInt64 rows = 0;
/// Size of files in filesystem
UInt64 bytes_compressed_on_disk = 0;
/// Makes sense for merges and mutations.
Strings source_part_names;
UInt64 bytes_uncompressed = 0;
UInt64 rows_read = 0;
UInt64 bytes_read_uncompressed = 0;
UInt64 peak_memory_usage = 0;
/// Was the operation successful?
UInt16 error = 0;
String exception;
std::shared_ptr<ProfileEvents::Counters::Snapshot> profile_counters;
static std::string name() { return "PartLog"; }
static MergeReasonType getMergeReasonType(MergeType merge_type);
static PartMergeAlgorithm getMergeAlgorithm(MergeAlgorithm merge_algorithm_);
static NamesAndTypesList getNamesAndTypes();
static NamesAndAliases getNamesAndAliases();
void appendToBlock(MutableColumns & columns) const;
static const char * getCustomColumnList() { return nullptr; }
};
class IMergeTreeDataPart;
/// Instead of typedef - to allow forward declaration.
class PartLog : public SystemLog<PartLogElement>
{
using SystemLog<PartLogElement>::SystemLog;
using MutableDataPartPtr = std::shared_ptr<IMergeTreeDataPart>;
using MutableDataPartsVector = std::vector<MutableDataPartPtr>;
using ProfileCountersSnapshotPtr = std::shared_ptr<ProfileEvents::Counters::Snapshot>;
public:
struct PartLogEntry
{
std::shared_ptr<IMergeTreeDataPart> part;
ProfileCountersSnapshotPtr profile_counters;
UInt64 elapsed_ns;
PartLogEntry(std::shared_ptr<IMergeTreeDataPart> part_, UInt64 elapsed_ns_)
: part(std::move(part_)), elapsed_ns(elapsed_ns_)
{
}
PartLogEntry(std::shared_ptr<IMergeTreeDataPart> part_, UInt64 elapsed_ns_, ProfileCountersSnapshotPtr profile_counters_)
: part(std::move(part_))
, profile_counters(std::move(profile_counters_))
, elapsed_ns(elapsed_ns_)
{
}
};
using PartLogEntries = std::vector<PartLogEntry>;
static PartLogEntries createPartLogEntries(const MutableDataPartsVector & parts, UInt64 elapsed_ns, ProfileCountersSnapshotPtr profile_counters = {});
/// Add a record about creation of new part.
static bool addNewPart(ContextPtr context, const PartLogEntry & part,
const ExecutionStatus & execution_status = {});
static bool addNewParts(ContextPtr context, const PartLogEntries & parts,
const ExecutionStatus & execution_status = {});
};
}
|
