1#pragma once
2
3#include "Algorithms.h"
4#include "ArraySourceVisitor.h"
5#include "ArraySinkVisitor.h"
6#include "ValueSourceVisitor.h"
7
8
9namespace DB::GatherUtils
10{
11/// Base classes which selects template function implementation with concrete ArraySource or ArraySink
12/// Derived classes should implement selectImpl for ArraySourceSelector and ArraySinkSelector,
13/// selectSourceSink for ArraySinkSourceSelector and selectSourcePair for ArraySourcePairSelector
14
15template <typename Base, typename Tuple, int index, typename ... Args>
16void callSelectMemberFunctionWithTupleArgument(Tuple & tuple, Args && ... args)
17{
18 if constexpr (index == std::tuple_size<Tuple>::value)
19 Base::selectImpl(args ...);
20 else
21 callSelectMemberFunctionWithTupleArgument<Base, Tuple, index + 1>(tuple, args ..., std::get<index>(tuple));
22}
23
24template <typename Base, typename ... Args>
25struct ArraySourceSelectorVisitor : public ArraySourceVisitorImpl<ArraySourceSelectorVisitor<Base, Args ...>>
26{
27 explicit ArraySourceSelectorVisitor(Args && ... args) : packed_args(args ...) {}
28
29 using Tuple = std::tuple<Args && ...>;
30
31 template <typename Source>
32 void visitImpl(Source & source)
33 {
34 callSelectMemberFunctionWithTupleArgument<Base, Tuple, 0>(packed_args, source);
35 }
36
37 Tuple packed_args;
38};
39
40template <typename Base>
41struct ArraySourceSelector
42{
43 template <typename ... Args>
44 static void select(IArraySource & source, Args && ... args)
45 {
46 ArraySourceSelectorVisitor<Base, Args ...> visitor(args ...);
47 source.accept(visitor);
48 }
49};
50
51
52template <typename Base, typename ... Args>
53struct ArraySinkSelectorVisitor : public ArraySinkVisitorImpl<ArraySinkSelectorVisitor<Base, Args ...>>
54{
55 explicit ArraySinkSelectorVisitor(Args && ... args) : packed_args(args ...) {}
56
57 using Tuple = std::tuple<Args && ...>;
58
59 template <typename Sink>
60 void visitImpl(Sink & sink)
61 {
62 callSelectMemberFunctionWithTupleArgument<Base, Tuple, 0>(packed_args, sink);
63 }
64
65 Tuple packed_args;
66};
67
68template <typename Base>
69struct ArraySinkSelector
70{
71 template <typename ... Args>
72 static void select(IArraySink & sink, Args && ... args)
73 {
74 ArraySinkSelectorVisitor<Base, Args ...> visitor(args ...);
75 sink.accept(visitor);
76 }
77};
78
79
80template <typename Base, typename ... Args>
81struct ValueSourceSelectorVisitor : public ValueSourceVisitorImpl<ValueSourceSelectorVisitor<Base, Args ...>>
82{
83 explicit ValueSourceSelectorVisitor(Args && ... args) : packed_args(args ...) {}
84
85 using Tuple = std::tuple<Args && ...>;
86
87 template <typename Source>
88 void visitImpl(Source & source)
89 {
90 callSelectMemberFunctionWithTupleArgument<Base, Tuple, 0>(packed_args, source);
91 }
92
93 Tuple packed_args;
94};
95
96template <typename Base>
97struct ValueSourceSelector
98{
99 template <typename ... Args>
100 static void select(IValueSource & source, Args && ... args)
101 {
102 ValueSourceSelectorVisitor<Base, Args ...> visitor(args ...);
103 source.accept(visitor);
104 }
105};
106
107template <typename Base>
108struct ArraySinkSourceSelector
109{
110 template <typename ... Args>
111 static void select(IArraySource & source, IArraySink & sink, Args && ... args)
112 {
113 ArraySinkSelector<Base>::select(sink, source, args ...);
114 }
115
116 template <typename Sink, typename ... Args>
117 static void selectImpl(Sink && sink, IArraySource & source, Args && ... args)
118 {
119 ArraySourceSelector<Base>::select(source, sink, args ...);
120 }
121
122 template <typename Source, typename Sink, typename ... Args>
123 static void selectImpl(Source && source, Sink && sink, Args && ... args)
124 {
125 Base::selectSourceSink(source, sink, args ...);
126 }
127};
128
129template <typename Base>
130struct ArraySourcePairSelector
131{
132 template <typename ... Args>
133 static void select(IArraySource & first, IArraySource & second, Args && ... args)
134 {
135 ArraySourceSelector<Base>::select(first, second, args ...);
136 }
137
138 template <typename FirstSource, typename ... Args>
139 static void selectImpl(FirstSource && first, IArraySource & second, Args && ... args)
140 {
141 ArraySourceSelector<Base>::select(second, first, args ...);
142 }
143
144 template <typename SecondSource, typename FirstSource, typename ... Args>
145 static void selectImpl(SecondSource && second, FirstSource && first, Args && ... args)
146 {
147 Base::selectSourcePair(first, second, args ...);
148 }
149};
150
151template <typename Base>
152struct ArrayAndValueSourceSelectorBySink : public ArraySinkSelector<ArrayAndValueSourceSelectorBySink<Base>>
153{
154 template <typename Sink, typename ... Args>
155 static void selectImpl(Sink && sink, IArraySource & array_source, IValueSource & value_source, Args && ... args)
156 {
157 using SynkType = typename std::decay<Sink>::type;
158 using ArraySource = typename SynkType::CompatibleArraySource;
159 using ValueSource = typename SynkType::CompatibleValueSource;
160
161 auto checkType = [] (auto source_ptr)
162 {
163 if (source_ptr == nullptr)
164 throw Exception(demangle(typeid(Base).name()) + " expected "
165 + demangle(typeid(typename SynkType::CompatibleArraySource).name())
166 + " or " + demangle(typeid(ConstSource<typename SynkType::CompatibleArraySource>).name())
167 + " or " + demangle(typeid(typename SynkType::CompatibleValueSource).name()) +
168 + " or " + demangle(typeid(ConstSource<typename SynkType::CompatibleValueSource>).name())
169 + " but got " + demangle(typeid(*source_ptr).name()), ErrorCodes::LOGICAL_ERROR);
170 };
171 auto checkTypeAndCallConcat = [& sink, & checkType, & args ...] (auto array_source_ptr, auto value_source_ptr)
172 {
173 checkType(array_source_ptr);
174 checkType(value_source_ptr);
175
176 Base::selectArrayAndValueSourceBySink(*array_source_ptr, *value_source_ptr, sink, args ...);
177 };
178
179 if (array_source.isConst() && value_source.isConst())
180 checkTypeAndCallConcat(typeid_cast<ConstSource<ArraySource> *>(&array_source),
181 typeid_cast<ConstSource<ValueSource> *>(&value_source));
182 else if (array_source.isConst())
183 checkTypeAndCallConcat(typeid_cast<ConstSource<ArraySource> *>(&array_source),
184 typeid_cast<ValueSource *>(&value_source));
185 else if (value_source.isConst())
186 checkTypeAndCallConcat(typeid_cast<ArraySource *>(&array_source),
187 typeid_cast<ConstSource<ValueSource> *>(&value_source));
188 else
189 checkTypeAndCallConcat(typeid_cast<ArraySource *>(&array_source),
190 typeid_cast<ValueSource *>(&value_source));
191 }
192};
193
194}
195