为Clickhouse增加自定义函数

在/data/apps/clickhouse-source/ClickHouse/dbms/src/AggregateFunctions目录下创建两个文件AggregateFunctionSequenceMatch1.h和AggregateFunctionSequenceMatch1.cpp

1 AggregateFunctionSequenceMatch1.h的代码如下:

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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
#pragma once

#include <AggregateFunctions/IAggregateFunction.h>
#include <DataTypes/DataTypeDateTime.h>
#include <DataTypes/DataTypesNumber.h>
#include <Columns/ColumnsNumber.h>
#include <Common/assert_cast.h>
#include <ext/range.h>
#include <Common/PODArray.h>
#include <IO/ReadHelpers.h>
#include <IO/WriteHelpers.h>
#include <bitset>
#include <stack>
#include <common/logger_useful.h>

namespace DB
{

namespace ErrorCodes
{
extern const int TOO_SLOW;
extern const int TOO_FEW_ARGUMENTS_FOR_FUNCTION;
extern const int TOO_MANY_ARGUMENTS_FOR_FUNCTION;
extern const int SYNTAX_ERROR;
extern const int BAD_ARGUMENTS;
extern const int LOGICAL_ERROR;
}

/// helper type for comparing `std::pair`s using solely the .first member
template <template <typename> class Comparator>
struct ComparePairFirst final
{
template <typename T1, typename T2>
bool operator()(const std::pair<T1, T2> & lhs, const std::pair<T1, T2> & rhs) const
{
return Comparator<T1>{}(lhs.first, rhs.first);
}
};

static constexpr auto max_events = 32;

template <typename T>
struct AggregateFunctionSequenceMatch1Data final
{
using Timestamp = T;
using Events = std::bitset<max_events>;
using TimestampEvents = std::pair<Timestamp, Events>;
using Comparator = ComparePairFirst<std::less>;

bool sorted = true;
PODArrayWithStackMemory<TimestampEvents, 64> events_list;

void add(const Timestamp timestamp, const Events & events)
{
/// store information exclusively for rows with at least one event
if (events.any())
{
events_list.emplace_back(timestamp, events);
sorted = false;
}
}

void merge(const AggregateFunctionSequenceMatch1Data & other)
{
if (other.events_list.empty())
return;

const auto size = events_list.size();

events_list.insert(std::begin(other.events_list), std::end(other.events_list));

/// either sort whole container or do so partially merging ranges afterwards
if (!sorted && !other.sorted)
std::sort(std::begin(events_list), std::end(events_list), Comparator{});
else
{
const auto begin = std::begin(events_list);
const auto middle = std::next(begin, size);
const auto end = std::end(events_list);

if (!sorted)
std::sort(begin, middle, Comparator{});

if (!other.sorted)
std::sort(middle, end, Comparator{});

std::inplace_merge(begin, middle, end, Comparator{});
}

sorted = true;
}

void sort()
{
if (!sorted)
{
std::sort(std::begin(events_list), std::end(events_list), Comparator{});
sorted = true;
}
}

void serialize(WriteBuffer & buf) const
{
writeBinary(sorted, buf);
writeBinary(events_list.size(), buf);

for (const auto & events : events_list)
{
writeBinary(events.first, buf);
writeBinary(events.second.to_ulong(), buf);
}
}

void deserialize(ReadBuffer & buf)
{
readBinary(sorted, buf);

size_t size;
readBinary(size, buf);

events_list.clear();
events_list.reserve(size);

for (size_t i = 0; i < size; ++i)
{
Timestamp timestamp;
readBinary(timestamp, buf);

UInt64 events;
readBinary(events, buf);

events_list.emplace_back(timestamp, Events{events});
}
}
};


/// Max number of iterations to match the pattern against a sequence, exception thrown when exceeded
constexpr auto sequence_match_max_iterations = 1000000;


template <typename T, typename Data, typename Derived>
class AggregateFunctionSequenceBase : public IAggregateFunctionDataHelper<Data, Derived>
{
public:
AggregateFunctionSequenceBase(const DataTypes & arguments, const Array & params, const String & pattern_)
: IAggregateFunctionDataHelper<Data, Derived>(arguments, params)
, pattern(pattern_)
{
arg_count = arguments.size();
parsePattern();
}

void add(AggregateDataPtr place, const IColumn ** columns, const size_t row_num, Arena *) const override
{
const auto timestamp = assert_cast<const ColumnVector<T> *>(columns[0])->getData()[row_num];

typename Data::Events events;
for (const auto i : ext::range(1, arg_count))
{
const auto event = assert_cast<const ColumnUInt8 *>(columns[i])->getData()[row_num];
events.set(i - 1, event);
}

this->data(place).add(timestamp, events);
}

void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
{
this->data(place).merge(this->data(rhs));
}

void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
{
this->data(place).serialize(buf);
}

void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
{
this->data(place).deserialize(buf);
}

const char * getHeaderFilePath() const override { return __FILE__; }

private:
enum class PatternActionType
{
SpecificEvent,
AnyEvent,
KleeneStar,
TimeLessOrEqual,
TimeLess,
TimeGreaterOrEqual,
TimeGreater
};

struct PatternAction final
{
PatternActionType type;
std::uint64_t extra;

PatternAction() = default;
PatternAction(const PatternActionType type_, const std::uint64_t extra_ = 0) : type{type_}, extra{extra_} {}
};

using PatternActions = PODArrayWithStackMemory<PatternAction, 64>;

Derived & derived() { return static_cast<Derived &>(*this); }

void parsePattern()
{
actions.clear();
actions.emplace_back(PatternActionType::KleeneStar);

dfa_states.clear();
dfa_states.emplace_back(true);

pattern_has_time = false;

const char * pos = pattern.data();
const char * begin = pos;
const char * end = pos + pattern.size();

auto throw_exception = [&](const std::string & msg)
{
throw Exception{msg + " '" + std::string(pos, end) + "' at position " + toString(pos - begin), ErrorCodes::SYNTAX_ERROR};
};

auto match = [&pos, end](const char * str) mutable
{
size_t length = strlen(str);
if (pos + length <= end && 0 == memcmp(pos, str, length))
{
pos += length;
return true;
}
return false;
};

while (pos < end)
{
if (match("(?"))
{
if (match("t"))
{
PatternActionType type;

if (match("<="))
type = PatternActionType::TimeLessOrEqual;
else if (match("<"))
type = PatternActionType::TimeLess;
else if (match(">="))
type = PatternActionType::TimeGreaterOrEqual;
else if (match(">"))
type = PatternActionType::TimeGreater;
else
throw_exception("Unknown time condition");

UInt64 duration = 0;
auto prev_pos = pos;
pos = tryReadIntText(duration, pos, end);
if (pos == prev_pos)
throw_exception("Could not parse number");

if (actions.back().type != PatternActionType::SpecificEvent &&
actions.back().type != PatternActionType::AnyEvent &&
actions.back().type != PatternActionType::KleeneStar)
throw Exception{"Temporal condition should be preceeded by an event condition", ErrorCodes::BAD_ARGUMENTS};

pattern_has_time = true;
actions.emplace_back(type, duration);
}
else
{
UInt64 event_number = 0;
auto prev_pos = pos;
pos = tryReadIntText(event_number, pos, end);
if (pos == prev_pos)
throw_exception("Could not parse number");

if (event_number > arg_count - 1)
throw Exception{"Event number " + toString(event_number) + " is out of range", ErrorCodes::BAD_ARGUMENTS};

actions.emplace_back(PatternActionType::SpecificEvent, event_number - 1);
dfa_states.back().transition = DFATransition::SpecificEvent;
dfa_states.back().event = event_number - 1;
dfa_states.emplace_back();
}

if (!match(")"))
throw_exception("Expected closing parenthesis, found");

}
else if (match(".*"))
{
actions.emplace_back(PatternActionType::KleeneStar);
dfa_states.back().has_kleene = true;
}
else if (match("."))
{
actions.emplace_back(PatternActionType::AnyEvent);
dfa_states.back().transition = DFATransition::AnyEvent;
dfa_states.emplace_back();
}
else
throw_exception("Could not parse pattern, unexpected starting symbol");
}
}

protected:
/// Uses a DFA based approach in order to better handle patterns without
/// time assertions.
///
/// NOTE: This implementation relies on the assumption that the pattern are *small*.
///
/// This algorithm performs in O(mn) (with m the number of DFA states and N the number
/// of events) with a memory consumption and memory allocations in O(m). It means that
/// if n >>> m (which is expected to be the case), this algorithm can be considered linear.
template <typename EventEntry>
bool dfaMatch(EventEntry & events_it, const EventEntry events_end) const
{
using ActiveStates = std::vector<bool>;

/// Those two vectors keep track of which states should be considered for the current
/// event as well as the states which should be considered for the next event.
ActiveStates active_states(dfa_states.size(), false);
ActiveStates next_active_states(dfa_states.size(), false);
active_states[0] = true;

/// Keeps track of dead-ends in order not to iterate over all the events to realize that
/// the match failed.
size_t n_active = 1;

for (/* empty */; events_it != events_end && n_active > 0 && !active_states.back(); ++events_it)
{
n_active = 0;
next_active_states.assign(dfa_states.size(), false);

for (size_t state = 0; state < dfa_states.size(); ++state)
{
if (!active_states[state])
{
continue;
}

switch (dfa_states[state].transition)
{
case DFATransition::None:
break;
case DFATransition::AnyEvent:
next_active_states[state + 1] = true;
++n_active;
break;
case DFATransition::SpecificEvent:
if (events_it->second.test(dfa_states[state].event))
{
next_active_states[state + 1] = true;
++n_active;
}
break;
}

if (dfa_states[state].has_kleene)
{
next_active_states[state] = true;
++n_active;
}
}
swap(active_states, next_active_states);
}

return active_states.back();
}

Logger * log = &Logger::get("AggregateFunctionSequenceMatch1");

template <typename EventEntry>
bool backtrackingMatch(EventEntry & events_it, const EventEntry events_end) const
{
const auto action_begin = std::begin(actions);
const auto action_end = std::end(actions);
auto action_it = action_begin;

const auto events_begin = events_it;
auto base_it = events_it;

/// an iterator to action plus an iterator to row in events list plus timestamp at the start of sequence
using backtrack_info = std::tuple<decltype(action_it), EventEntry, EventEntry>;
std::stack<backtrack_info> back_stack;

/// backtrack if possible
const auto do_backtrack = [&]
{
while (!back_stack.empty())
{
auto & top = back_stack.top();

action_it = std::get<0>(top);
events_it = std::next(std::get<1>(top));
base_it = std::get<2>(top);

back_stack.pop();

if (events_it != events_end)
return true;
}

return false;
};

size_t i = 0;
while (action_it != action_end && events_it != events_end)
{
if (action_it->type == PatternActionType::SpecificEvent)
{
LOG_DEBUG(log, "hello here SequenceMatch1 " << std::endl );

if (events_it->second.test(action_it->extra))
{
/// move to the next action and events
base_it = events_it;
++action_it, ++events_it;
}
else if (!do_backtrack())
/// backtracking failed, bail out
break;
}
else if (action_it->type == PatternActionType::AnyEvent)
{
base_it = events_it;
++action_it, ++events_it;
}
else if (action_it->type == PatternActionType::KleeneStar)
{
back_stack.emplace(action_it, events_it, base_it);
base_it = events_it;
++action_it;
}
else if (action_it->type == PatternActionType::TimeLessOrEqual)
{
if (events_it->first <= base_it->first + action_it->extra)
{
/// condition satisfied, move onto next action
back_stack.emplace(action_it, events_it, base_it);
base_it = events_it;
++action_it;
}
else if (!do_backtrack())
break;
}
else if (action_it->type == PatternActionType::TimeLess)
{
if (events_it->first < base_it->first + action_it->extra)
{
back_stack.emplace(action_it, events_it, base_it);
base_it = events_it;
++action_it;
}
else if (!do_backtrack())
break;
}
else if (action_it->type == PatternActionType::TimeGreaterOrEqual)
{
if (events_it->first >= base_it->first + action_it->extra)
{
back_stack.emplace(action_it, events_it, base_it);
base_it = events_it;
++action_it;
}
else if (++events_it == events_end && !do_backtrack())
break;
}
else if (action_it->type == PatternActionType::TimeGreater)
{
if (events_it->first > base_it->first + action_it->extra)
{
back_stack.emplace(action_it, events_it, base_it);
LOG_DEBUG(log, "action_it: " << action_it->extra << " base_it: " << base_it->first << " events_it: " << events_it->first << std::endl );
LOG_DEBUG(log, "action_it: " << action_it->extra << " base_it: " << base_it->first << " events_it: " << events_it->first << std::endl );
base_it = events_it;

++action_it;
}
else if (++events_it == events_end && !do_backtrack())
break;
}
else
throw Exception{"Unknown PatternActionType", ErrorCodes::LOGICAL_ERROR};

if (++i > sequence_match_max_iterations)
throw Exception{"Pattern application proves too difficult, exceeding max iterations (" + toString(sequence_match_max_iterations) + ")",
ErrorCodes::TOO_SLOW};
}

/// if there are some actions remaining
if (action_it != action_end)
{
/// match multiple empty strings at end
while (action_it->type == PatternActionType::KleeneStar ||
action_it->type == PatternActionType::TimeLessOrEqual ||
action_it->type == PatternActionType::TimeLess ||
(action_it->type == PatternActionType::TimeGreaterOrEqual && action_it->extra == 0))
++action_it;
}

if (events_it == events_begin)
{
++events_it;
LOG_DEBUG(log, "++events_it; " << std::endl );
}

if(action_it == action_end){
--events_it;
}

return action_it == action_end;
}

private:
enum class DFATransition : char
{
/// .-------.
/// | |
/// `-------'
None,
/// .-------. (?[0-9])
/// | | ----------
/// `-------'
SpecificEvent,
/// .-------. .
/// | | ----------
/// `-------'
AnyEvent,
};

struct DFAState
{
DFAState(bool has_kleene_ = false)
: has_kleene{has_kleene_}, event{0}, transition{DFATransition::None}
{}

/// .-------.
/// | | - - -
/// `-------'
/// |_^
bool has_kleene;
/// In the case of a state transitions with a `SpecificEvent`,
/// `event` contains the value of the event.
uint32_t event;
/// The kind of transition out of this state.
DFATransition transition;
};

using DFAStates = std::vector<DFAState>;

protected:
/// `True` if the parsed pattern contains time assertions (?t...), `false` otherwise.
bool pattern_has_time;

private:
std::string pattern;
size_t arg_count;
PatternActions actions;

DFAStates dfa_states;
};

template <typename T, typename Data>
class AggregateFunctionSequenceMatch1 final : public AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceMatch1<T, Data>>
{
public:
AggregateFunctionSequenceMatch1(const DataTypes & arguments, const Array & params, const String & pattern_)
: AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceMatch1<T, Data>>(arguments, params, pattern_) {}

using AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceMatch1<T, Data>>::AggregateFunctionSequenceBase;

String getName() const override { return "sequenceMatch"; }

DataTypePtr getReturnType() const override { return std::make_shared<DataTypeUInt8>(); }

void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
{
const_cast<Data &>(this->data(place)).sort();

const auto & data_ref = this->data(place);

const auto events_begin = std::begin(data_ref.events_list);
const auto events_end = std::end(data_ref.events_list);
auto events_it = events_begin;

bool match = this->pattern_has_time ? this->backtrackingMatch(events_it, events_end) : this->dfaMatch(events_it, events_end);
assert_cast<ColumnUInt8 &>(to).getData().push_back(match);
}
};

template <typename T, typename Data>
class AggregateFunctionSequenceCount1 final : public AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceCount1<T, Data>>
{
public:
AggregateFunctionSequenceCount1(const DataTypes & arguments, const Array & params, const String & pattern_)
: AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceCount1<T, Data>>(arguments, params, pattern_) {}

using AggregateFunctionSequenceBase<T, Data, AggregateFunctionSequenceCount1<T, Data>>::AggregateFunctionSequenceBase;

String getName() const override { return "sequenceCount"; }

DataTypePtr getReturnType() const override { return std::make_shared<DataTypeUInt64>(); }

void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
{
const_cast<Data &>(this->data(place)).sort();
assert_cast<ColumnUInt64 &>(to).getData().push_back(count(place));
}

private:
UInt64 count(const ConstAggregateDataPtr & place) const
{
const auto & data_ref = this->data(place);

const auto events_begin = std::begin(data_ref.events_list);
const auto events_end = std::end(data_ref.events_list);
auto events_it = events_begin;

size_t count = 0;
while (events_it != events_end && this->backtrackingMatch(events_it, events_end))
++count;

return count;
}
};

}

AggregateFunctionSequenceMatch1.cpp的代码如下:

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
#include <AggregateFunctions/Helpers.h>
#include <AggregateFunctions/AggregateFunctionFactory.h>
#include <AggregateFunctions/AggregateFunctionSequenceMatch1.h>

#include <DataTypes/DataTypeDate.h>
#include <DataTypes/DataTypeDateTime.h>

#include <ext/range.h>

namespace DB
{

namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}

namespace
{

template <template <typename, typename> class AggregateFunction, template <typename> class Data>
AggregateFunctionPtr createAggregateFunctionSequenceBase(const std::string & name, const DataTypes & argument_types, const Array & params)
{
if (params.size() != 1)
throw Exception{"Aggregate function " + name + " requires exactly one parameter.",
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH};

const auto arg_count = argument_types.size();

if (arg_count < 3)
throw Exception{"Aggregate function " + name + " requires at least 3 arguments.",
ErrorCodes::TOO_FEW_ARGUMENTS_FOR_FUNCTION};

if (arg_count - 1 > max_events)
throw Exception{"Aggregate function " + name + " supports up to "
+ toString(max_events) + " event arguments.",
ErrorCodes::TOO_MANY_ARGUMENTS_FOR_FUNCTION};

const auto time_arg = argument_types.front().get();

for (const auto i : ext::range(1, arg_count))
{
const auto cond_arg = argument_types[i].get();
if (!isUInt8(cond_arg))
throw Exception{"Illegal type " + cond_arg->getName() + " of argument " + toString(i + 1)
+ " of aggregate function " + name + ", must be UInt8",
ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT};
}

String pattern = params.front().safeGet<std::string>();

AggregateFunctionPtr res(createWithUnsignedIntegerType<AggregateFunction, Data>(*argument_types[0], argument_types, params, pattern));
if (res)
return res;

WhichDataType which(argument_types.front().get());
if (which.isDateTime())
return std::make_shared<AggregateFunction<DataTypeDateTime::FieldType, Data<DataTypeDateTime::FieldType>>>(argument_types, params, pattern);
else if (which.isDate())
return std::make_shared<AggregateFunction<DataTypeDate::FieldType, Data<DataTypeDate::FieldType>>>(argument_types, params, pattern);

throw Exception{"Illegal type " + time_arg->getName() + " of first argument of aggregate function "
+ name + ", must be DateTime",
ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT};
}

}

void registerAggregateFunctionsSequenceMatch1(AggregateFunctionFactory & factory)
{
factory.registerFunction("sequenceMatch1", createAggregateFunctionSequenceBase<AggregateFunctionSequenceMatch1, AggregateFunctionSequenceMatch1Data>);
factory.registerFunction("sequenceCount1", createAggregateFunctionSequenceBase<AggregateFunctionSequenceCount1, AggregateFunctionSequenceMatch1Data>);
}

}

2 在/data/apps/clickhouse-source/ClickHouse/dbms/src/AggregateFunctions/registerAggregateFunctions.cpp中添加两行代码

1
2
registerAggregateFunctions.cpp:17:void registerAggregateFunctionsSequenceMatch1(AggregateFunctionFactory &);
registerAggregateFunctions.cpp:60: registerAggregateFunctionsSequenceMatch1(factory);

3 在build目录执行

1
2
cmake ..
ninja clickhouse

之后重启Clickhouse服务端

分享