Project
Loading...
Searching...
No Matches
AnalysisTask.h
Go to the documentation of this file.
1// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3// All rights not expressly granted are reserved.
4//
5// This software is distributed under the terms of the GNU General Public
6// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7//
8// In applying this license CERN does not waive the privileges and immunities
9// granted to it by virtue of its status as an Intergovernmental Organization
10// or submit itself to any jurisdiction.
11
12#ifndef FRAMEWORK_ANALYSIS_TASK_H_
13#define FRAMEWORK_ANALYSIS_TASK_H_
14
25#include "Framework/Traits.h"
29
30#include <arrow/compute/kernel.h>
31#include <arrow/table.h>
32#include <gandiva/node.h>
33#include <type_traits>
34#include <utility>
35#include <memory>
36namespace o2::framework
37{
46};
47
48template <int64_t BEGIN, int64_t END, int64_t STEP = 1>
50 static constexpr int64_t begin = BEGIN;
51 static constexpr int64_t end = END;
52 static constexpr int64_t step = STEP;
53};
54
55template <typename T>
56static constexpr bool is_enumeration_v = false;
57
58template <int64_t BEGIN, int64_t END, int64_t STEP>
59static constexpr bool is_enumeration_v<Enumeration<BEGIN, END, STEP>> = true;
60
61template <typename T>
62concept is_enumeration = is_enumeration_v<std::decay_t<T>>;
63
64// Helper struct which builds a DataProcessorSpec from
65// the contents of an AnalysisTask...
66namespace {
67struct AnalysisDataProcessorBuilder {
68 template <typename G, typename... Args>
69 static void addGroupingCandidates(std::vector<StringPair>& bk, std::vector<StringPair>& bku)
70 {
71 [&bk, &bku]<typename... As>(framework::pack<As...>) mutable {
72 std::string key;
73 if constexpr (soa::is_iterator<std::decay_t<G>>) {
74 key = std::string{"fIndex"} + o2::framework::cutString(soa::getLabelFromType<std::decay_t<G>>());
75 }
76 ([&bk, &bku, &key]() mutable {
77 if constexpr (soa::relatedByIndex<std::decay_t<G>, std::decay_t<As>>()) {
78 auto binding = soa::getLabelFromTypeForKey<std::decay_t<As>>(key);
80 framework::updatePairList(bku, binding, key);
81 } else {
82 framework::updatePairList(bk, binding, key);
83 }
84 }
85 }(),
86 ...);
87 }(framework::pack<Args...>{});
88 }
89
90 template <soa::TableRef R>
91 static void addOriginalRef(const char* name, bool value, std::vector<InputSpec>& inputs)
92 {
93 auto spec = soa::tableRef2InputSpec<R>();
94 spec.metadata.emplace_back(ConfigParamSpec{std::string{"control:"} + name, VariantType::Bool, value, {"\"\""}});
95 DataSpecUtils::updateInputList(inputs, std::move(spec));
96 }
97
99 template <soa::is_table A>
101 static void addExpression(int, uint32_t, std::vector<ExpressionInfo>&)
102 {
103 }
104
105 template <soa::is_filtered_table A>
106 static void addExpression(int ai, uint32_t hash, std::vector<ExpressionInfo>& eInfos)
107 {
108 auto fields = soa::createFieldsFromColumns(typename std::decay_t<A>::persistent_columns_t{});
109 eInfos.emplace_back(ai, hash, std::decay_t<A>::hashes(), std::make_shared<arrow::Schema>(fields));
110 }
111
112 template <soa::is_iterator A>
113 static void addExpression(int ai, uint32_t hash, std::vector<ExpressionInfo>& eInfos)
114 {
115 addExpression<typename std::decay_t<A>::parent_t>(ai, hash, eInfos);
116 }
117
119 template <soa::is_table A>
120 static void addInput(const char* name, bool value, std::vector<InputSpec>& inputs)
121 {
122 [&name, &value, &inputs]<size_t N, std::array<soa::TableRef, N> refs, size_t... Is>(std::index_sequence<Is...>) mutable {
123 (addOriginalRef<refs[Is]>(name, value, inputs), ...);
124 }.template operator()<A::originals.size(), std::decay_t<A>::originals>(std::make_index_sequence<std::decay_t<A>::originals.size()>());
125 }
126
127 template <soa::is_iterator A>
128 static void addInput(const char* name, bool value, std::vector<InputSpec>& inputs)
129 {
130 addInput<typename std::decay_t<A>::parent_t>(name, value, inputs);
131 }
132
134 template <soa::is_table... As>
135 static void addInputsAndExpressions(uint32_t hash, const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos)
136 {
137 int ai = -1;
138 ([&ai, &hash, &eInfos, &name, &value, &inputs]() mutable {
139 ++ai;
140 using T = std::decay_t<As>;
141 addExpression<T>(ai, hash, eInfos);
142 addInput<T>(name, value, inputs);
143 }(),
144 ...);
145 }
146
149 template <typename R, typename C, is_enumeration A>
150 static void inputsFromArgs(R (C::*)(A), const char* /*name*/, bool /*value*/, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>&, std::vector<StringPair>&, std::vector<StringPair>&)
151 {
152 std::vector<ConfigParamSpec> inputMetadata;
153 // FIXME: for the moment we do not support begin, end and step.
154 DataSpecUtils::updateInputList(inputs, InputSpec{"enumeration", "DPL", "ENUM", 0, Lifetime::Enumeration, inputMetadata});
155 }
156
158 template <typename R, typename C, soa::is_iterator A, soa::is_table... Args>
159 static void inputsFromArgs(R (C::*)(A, Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos, std::vector<StringPair>& bk, std::vector<StringPair>& bku)
160 requires(std::is_lvalue_reference_v<A> && (std::is_lvalue_reference_v<Args> && ...))
161 {
162 addGroupingCandidates<A, Args...>(bk, bku);
163 constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(A, Args...)>();
164 addInputsAndExpressions<typename std::decay_t<A>::parent_t, Args...>(hash, name, value, inputs, eInfos);
165 }
166
168 template <typename R, typename C, soa::is_table... Args>
169 static void inputsFromArgs(R (C::*)(Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos, std::vector<StringPair>&, std::vector<StringPair>&)
170 requires(std::is_lvalue_reference_v<Args> && ...)
171 {
172 constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(Args...)>();
173 addInputsAndExpressions<Args...>(hash, name, value, inputs, eInfos);
174 }
175
176 template <soa::TableRef R>
177 static auto extractTableFromRecord(InputRecord& record)
178 {
179 auto table = record.get<TableConsumer>(o2::aod::label<R>())->asArrowTable();
180 if (table->num_rows() == 0) {
181 table = makeEmptyTable<R>();
182 }
183 return table;
184 }
185
186 template <soa::is_table T>
187 static auto extractFromRecord(InputRecord& record)
188 {
189 return T { [&record]<size_t N, std::array<soa::TableRef, N> refs, size_t... Is>(std::index_sequence<Is...>) { return std::vector{extractTableFromRecord<refs[Is]>(record)...}; }.template operator()<T::originals.size(), T::originals>(std::make_index_sequence<T::originals.size()>()) };
190 }
191
192 template <soa::is_iterator T>
193 static auto extractFromRecord(InputRecord& record)
194 {
195 return typename T::parent_t { [&record]<size_t N, std::array<soa::TableRef, N> refs, size_t... Is>(std::index_sequence<Is...>) { return std::vector{extractTableFromRecord<refs[Is]>(record)...}; }.template operator()<T::parent_t::originals.size(), T::parent_t::originals>(std::make_index_sequence<T::parent_t::originals.size()>()) };
196 }
197
198 template <soa::is_filtered T>
199 static auto extractFilteredFromRecord(InputRecord& record, ExpressionInfo& info)
200 {
201 std::shared_ptr<arrow::Table> table = nullptr;
202 auto joiner = [&record]<size_t N, std::array<soa::TableRef, N> refs, size_t... Is>(std::index_sequence<Is...>) { return std::vector{extractTableFromRecord<refs[Is]>(record)...}; };
203 if constexpr (soa::is_iterator<T>) {
204 table = o2::soa::ArrowHelpers::joinTables(joiner.template operator()<T::parent_t::originals.size(), T::parent_t::originals>(std::make_index_sequence<T::parent_t::originals.size()>()));
205 } else {
206 table = o2::soa::ArrowHelpers::joinTables(joiner.template operator()<T::originals.size(), T::originals>(std::make_index_sequence<T::originals.size()>()));
207 }
210 if (info.selection == nullptr) {
212 }
213 }
214 if constexpr (soa::is_iterator<T>) {
215 return typename T::parent_t({table}, info.selection);
216 } else {
217 return T({table}, info.selection);
218 }
219 }
220
221 template <is_enumeration T, int AI>
222 static auto extract(InputRecord&, std::vector<ExpressionInfo>&, size_t)
223 {
224 return T{};
225 }
226
227 template <soa::is_iterator T, int AI>
228 static auto extract(InputRecord& record, std::vector<ExpressionInfo>& infos, size_t phash)
229 {
230 if constexpr (std::same_as<typename T::policy_t, soa::FilteredIndexPolicy>) {
231 return extractFilteredFromRecord<T>(record, *std::find_if(infos.begin(), infos.end(), [&phash](ExpressionInfo const& i) { return (i.processHash == phash && i.argumentIndex == AI); }));
232 } else {
233 return extractFromRecord<T>(record);
234 }
235 }
236
237 template <soa::is_table T, int AI>
238 static auto extract(InputRecord& record, std::vector<ExpressionInfo>& infos, size_t phash)
239 {
240 if constexpr (soa::is_filtered_table<T>) {
241 return extractFilteredFromRecord<T>(record, *std::find_if(infos.begin(), infos.end(), [&phash](ExpressionInfo const& i) { return (i.processHash == phash && i.argumentIndex == AI); }));
242 } else {
243 return extractFromRecord<T>(record);
244 }
245 }
246
247 template <typename R, typename C, typename Grouping, typename... Args>
248 static auto bindGroupingTable(InputRecord& record, R (C::*)(Grouping, Args...), std::vector<ExpressionInfo>& infos)
249 requires(!std::same_as<Grouping, void>)
250 {
251 constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(Grouping, Args...)>();
252 return extract<std::decay_t<Grouping>, 0>(record, infos, hash);
253 }
254
255 template <typename R, typename C, typename Grouping, typename... Args>
256 static auto bindAssociatedTables(InputRecord& record, R (C::*)(Grouping, Args...), std::vector<ExpressionInfo>& infos)
257 requires(!std::same_as<Grouping, void> && sizeof...(Args) > 0)
258 {
259 constexpr auto p = pack<Args...>{};
260 constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(Grouping, Args...)>();
261 return std::make_tuple(extract<std::decay_t<Args>, has_type_at_v<Args>(p) + 1>(record, infos, hash)...);
262 }
263
264 template <typename... As>
265 static void overwriteInternalIndices(std::tuple<As...>& dest, std::tuple<As...> const& src)
266 {
267 (std::get<As>(dest).bindInternalIndicesTo(&std::get<As>(src)), ...);
268 }
269
270 template <typename Task, typename R, typename C, typename Grouping, typename... Associated>
271 static void invokeProcess(Task& task, InputRecord& inputs, R (C::*processingFunction)(Grouping, Associated...), std::vector<ExpressionInfo>& infos, ArrowTableSlicingCache& slices)
272 {
273 using G = std::decay_t<Grouping>;
274 auto groupingTable = AnalysisDataProcessorBuilder::bindGroupingTable(inputs, processingFunction, infos);
275
276 // set filtered tables for partitions with grouping
277 homogeneous_apply_refs([&groupingTable](auto& element) {
278 analysis_task_parsers::setPartition(element, groupingTable);
280 return true;
281 },
282 task);
283
284 if constexpr (sizeof...(Associated) == 0) {
285 // single argument to process
286 homogeneous_apply_refs([&groupingTable](auto& element) {
288 analysis_task_parsers::setGroupedCombination(element, groupingTable);
289 return true;
290 },
291 task);
292 if constexpr (soa::is_iterator<G>) {
293 for (auto& element : groupingTable) {
294 std::invoke(processingFunction, task, *element);
295 }
296 } else {
297 static_assert(soa::is_table<G> || is_enumeration<G>,
298 "Single argument of process() should be a table-like or an iterator");
299 std::invoke(processingFunction, task, groupingTable);
300 }
301 } else {
302 // multiple arguments to process
303 static_assert(((soa::is_iterator<std::decay_t<Associated>> == false) && ...),
304 "Associated arguments of process() should not be iterators");
305 auto associatedTables = AnalysisDataProcessorBuilder::bindAssociatedTables(inputs, processingFunction, infos);
306 // pre-bind self indices
307 std::apply(
308 [&task](auto&... t) mutable {
310 [&t](auto& p) {
312 return true;
313 },
314 task),
315 ...);
316 },
317 associatedTables);
318
319 auto binder = [&task, &groupingTable, &associatedTables](auto& x) mutable {
320 x.bindExternalIndices(&groupingTable, &std::get<std::decay_t<Associated>>(associatedTables)...);
321 homogeneous_apply_refs([&x](auto& t) mutable {
324 return true;
325 },
326 task);
327 };
328 groupingTable.bindExternalIndices(&std::get<std::decay_t<Associated>>(associatedTables)...);
329
330 // always pre-bind full tables to support index hierarchy
331 std::apply(
332 [&binder](auto&... x) mutable {
333 (binder(x), ...);
334 },
335 associatedTables);
336
337 // GroupedCombinations bound separately, as they should be set once for all associated tables
338 homogeneous_apply_refs([&groupingTable, &associatedTables](auto& t) {
339 analysis_task_parsers::setGroupedCombination(t, groupingTable, associatedTables);
340 return true;
341 },
342 task);
343 overwriteInternalIndices(associatedTables, associatedTables);
344 if constexpr (soa::is_iterator<std::decay_t<G>>) {
345 auto slicer = GroupSlicer(groupingTable, associatedTables, slices);
346 for (auto& slice : slicer) {
347 auto associatedSlices = slice.associatedTables();
348 overwriteInternalIndices(associatedSlices, associatedTables);
349 std::apply(
350 [&binder](auto&... x) mutable {
351 (binder(x), ...);
352 },
353 associatedSlices);
354
355 // bind partitions and grouping table
356 homogeneous_apply_refs([&groupingTable](auto& x) {
358 return true;
359 },
360 task);
361
362 invokeProcessWithArgs(task, processingFunction, slice.groupingElement(), associatedSlices);
363 }
364 } else {
365 // bind partitions and grouping table
366 homogeneous_apply_refs([&groupingTable](auto& x) {
368 return true;
369 },
370 task);
371
372 invokeProcessWithArgs(task, processingFunction, groupingTable, associatedTables);
373 }
374 }
375 }
376
377 template <typename C, typename T, typename G, typename... A>
378 static void invokeProcessWithArgs(C& task, T processingFunction, G g, std::tuple<A...>& at)
379 {
380 std::invoke(processingFunction, task, g, std::get<A>(at)...);
381 }
382};
383}
384
386 std::vector<std::pair<std::string, bool>> map;
387};
388
390struct TaskName {
391 TaskName(std::string name) : value{std::move(name)} {}
392 std::string value;
393};
394
395namespace {
396template <typename T, typename... A>
397auto getTaskNameSetProcesses(std::string& outputName, TaskName first, SetDefaultProcesses second, A... args)
398{
399 auto task = std::make_shared<T>(std::forward<A>(args)...);
400 for (auto& setting : second.map) {
402 [&](auto& element) {
403 return analysis_task_parsers::setProcessSwitch(setting, element);
404 },
405 *task.get());
406 }
407 outputName = first.value;
408 return task;
409}
410
411template <typename T, typename... A>
412auto getTaskNameSetProcesses(std::string& outputName, SetDefaultProcesses first, TaskName second, A... args)
413{
414 auto task = std::make_shared<T>(std::forward<A>(args)...);
415 for (auto& setting : first.map) {
417 [&](auto& element) {
418 return analysis_task_parsers::setProcessSwitch(setting, element);
419 },
420 *task.get());
421 }
422 outputName = second.value;
423 return task;
424}
425
426template <typename T, typename... A>
427auto getTaskNameSetProcesses(std::string& outputName, SetDefaultProcesses first, A... args)
428{
429 auto task = std::make_shared<T>(std::forward<A>(args)...);
430 for (auto& setting : first.map) {
432 [&](auto& element) {
433 return analysis_task_parsers::setProcessSwitch(setting, element);
434 },
435 *task.get());
436 }
437 auto type_name_str = type_name<T>();
438 outputName = type_to_task_name(type_name_str);
439 return task;
440}
441
442template <typename T, typename... A>
443auto getTaskNameSetProcesses(std::string& outputName, TaskName first, A... args)
444{
445 auto task = std::make_shared<T>(std::forward<A>(args)...);
446 outputName = first.value;
447 return task;
448}
449
450template <typename T, typename... A>
451auto getTaskNameSetProcesses(std::string& outputName, A... args)
452{
453 auto task = std::make_shared<T>(std::forward<A>(args)...);
454 auto type_name_str = type_name<T>();
455 outputName = type_to_task_name(type_name_str);
456 return task;
457}
458
459}
460
463template <typename T, typename... Args>
465{
466 TH1::AddDirectory(false);
467
468 std::string name_str;
469 auto task = getTaskNameSetProcesses<T>(name_str, args...);
470
471 auto suffix = ctx.options().get<std::string>("workflow-suffix");
472 if (!suffix.empty()) {
473 name_str += suffix;
474 }
475 const char* name = name_str.c_str();
476
477 auto hash = runtime_hash(name);
478
479 std::vector<OutputSpec> outputs;
480 std::vector<InputSpec> inputs;
481 std::vector<ConfigParamSpec> options;
482 std::vector<ExpressionInfo> expressionInfos;
483 std::vector<StringPair> bindingsKeys;
484 std::vector<StringPair> bindingsKeysUnsorted;
485
487 homogeneous_apply_refs([&options, &hash](auto& element) { return analysis_task_parsers::appendOption(options, element); }, *task.get());
489 homogeneous_apply_refs([&inputs](auto& element) { return analysis_task_parsers::appendCondition(inputs, element); }, *task.get());
490
492 if constexpr (requires { &T::process; }) {
493 AnalysisDataProcessorBuilder::inputsFromArgs(&T::process, "default", true, inputs, expressionInfos, bindingsKeys, bindingsKeysUnsorted);
494 }
497 [name = name_str, &expressionInfos, &inputs, &bindingsKeys, &bindingsKeysUnsorted](framework::is_process_configurable auto& x) mutable {
498 // this pushes (argumentIndex,processHash,schemaPtr,nullptr) into expressionInfos for arguments that are Filtered/filtered_iterators
499 AnalysisDataProcessorBuilder::inputsFromArgs(x.process, (name + "/" + x.name).c_str(), x.value, inputs, expressionInfos, bindingsKeys, bindingsKeysUnsorted);
500 return true;
501 },
502 [](auto&) {
503 return false;
504 }},
505 *task.get());
506
507 // add preslice declarations to slicing cache definition
508 homogeneous_apply_refs([&bindingsKeys, &bindingsKeysUnsorted](auto& element) { return analysis_task_parsers::registerCache(element, bindingsKeys, bindingsKeysUnsorted); }, *task.get());
509
510 // request base tables for spawnable extended tables and indices to be built
511 // this checks for duplications
512 homogeneous_apply_refs([&inputs](auto& element) {
513 return analysis_task_parsers::requestInputs(inputs, element);
514 },
515 *task.get());
516
517 // no static way to check if the task defines any processing, we can only make sure it subscribes to at least something
518 if (inputs.empty() == true) {
519 LOG(warn) << "Task " << name_str << " has no inputs";
520 }
521
522 homogeneous_apply_refs([&outputs, &hash](auto& element) { return analysis_task_parsers::appendOutput(outputs, element, hash); }, *task.get());
523
524 auto requiredServices = CommonServices::defaultServices();
525 auto arrowServices = CommonServices::arrowServices();
526 requiredServices.insert(requiredServices.end(), arrowServices.begin(), arrowServices.end());
527 homogeneous_apply_refs([&requiredServices](auto& element) { return analysis_task_parsers::addService(requiredServices, element); }, *task.get());
528
529 auto algo = AlgorithmSpec::InitCallback{[task = task, expressionInfos, bindingsKeys, bindingsKeysUnsorted](InitContext& ic) mutable {
530 homogeneous_apply_refs([&ic](auto&& element) { return analysis_task_parsers::prepareOption(ic, element); }, *task.get());
531 homogeneous_apply_refs([&ic](auto&& element) { return analysis_task_parsers::prepareService(ic, element); }, *task.get());
532
533 auto& callbacks = ic.services().get<CallbackService>();
534 auto eoscb = [task](EndOfStreamContext& eosContext) {
535 homogeneous_apply_refs([&eosContext](auto& element) {
536 analysis_task_parsers::postRunService(eosContext, element);
537 analysis_task_parsers::postRunOutput(eosContext, element);
538 return true; },
539 *task.get());
540 eosContext.services().get<ControlService>().readyToQuit(QuitRequest::Me);
541 };
542
543 callbacks.set<CallbackService::Id::EndOfStream>(eoscb);
544
547 [&ic](auto& element) -> bool { return analysis_task_parsers::updatePlaceholders(ic, element); },
548 *task.get());
550 homogeneous_apply_refs([&expressionInfos](auto& element) {
551 return analysis_task_parsers::createExpressionTrees(expressionInfos, element);
552 },
553 *task.get());
554
555 if constexpr (requires { task->init(ic); }) {
556 task->init(ic);
557 }
558
559 ic.services().get<ArrowTableSlicingCacheDef>().setCaches(std::move(bindingsKeys));
560 ic.services().get<ArrowTableSlicingCacheDef>().setCachesUnsorted(std::move(bindingsKeysUnsorted));
561 // initialize global caches
562 homogeneous_apply_refs([&ic](auto& element) {
564 },
565 *(task.get()));
566
567 return [task, expressionInfos](ProcessingContext& pc) mutable {
568 // load the ccdb object from their cache
569 homogeneous_apply_refs([&pc](auto& element) { return analysis_task_parsers::newDataframeCondition(pc.inputs(), element); }, *task.get());
570 // reset partitions once per dataframe
571 homogeneous_apply_refs([](auto& element) { return analysis_task_parsers::newDataframePartition(element); }, *task.get());
572 // reset selections for the next dataframe
573 for (auto& info : expressionInfos) {
574 info.resetSelection = true;
575 }
576 // reset pre-slice for the next dataframe
577 auto slices = pc.services().get<ArrowTableSlicingCache>();
578 homogeneous_apply_refs([&pc, &slices](auto& element) {
579 return analysis_task_parsers::updateSliceInfo(element, slices);
580 },
581 *(task.get()));
582 // initialize local caches
583 homogeneous_apply_refs([&pc](auto& element) { return analysis_task_parsers::initializeCache(pc, element); }, *(task.get()));
584 // prepare outputs
585 homogeneous_apply_refs([&pc](auto& element) { return analysis_task_parsers::prepareOutput(pc, element); }, *task.get());
586 // execute run()
587 if constexpr (requires { task->run(pc); }) {
588 task->run(pc);
589 }
590 // execute process()
591 if constexpr (requires { AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), &T::process, expressionInfos, slices); }) {
592 AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), &T::process, expressionInfos, slices);
593 }
594 // execute optional process()
596 [&pc, &expressionInfos, &task, &slices](auto& x) mutable {
597 if constexpr (base_of_template<ProcessConfigurable, std::decay_t<decltype(x)>>) {
598 if (x.value == true) {
599 AnalysisDataProcessorBuilder::invokeProcess(*task.get(), pc.inputs(), x.process, expressionInfos, slices);
600 return true;
601 }
602 }
603 return false;
604 },
605 *task.get());
606 // finalize outputs
607 homogeneous_apply_refs([&pc](auto& element) { return analysis_task_parsers::finalizeOutput(pc, element); }, *task.get());
608 };
609 }};
610
611 return {
612 name,
613 // FIXME: For the moment we hardcode this. We could build
614 // this list from the list of methods actually implemented in the
615 // task itself.
616 inputs,
617 outputs,
618 algo,
619 options,
620 requiredServices};
621}
622
623} // namespace o2::framework
624#endif // FRAMEWORK_ANALYSISTASK_H_
int32_t i
constexpr uint32_t runtime_hash(char const *str)
StringRef key
Definition A.h:16
ConfigParamRegistry & options() const
Helper to check if a type T is an iterator.
Definition ASoA.h:1254
GLint GLenum GLint x
Definition glcorearb.h:403
GLenum src
Definition glcorearb.h:1767
GLuint GLuint end
Definition glcorearb.h:469
GLuint const GLchar * name
Definition glcorearb.h:781
GLsizei const GLfloat * value
Definition glcorearb.h:819
GLboolean GLboolean g
Definition glcorearb.h:1233
bool prepareService(InitContext &, T &)
bool requestInputs(std::vector< InputSpec > &, T const &)
Table auto-creation handling.
void setGroupedCombination(C &, TG &, Ts &...)
Combinations handling.
bool initializeCache(ProcessingContext &, T &)
bool preInitializeCache(InitContext &, T &)
Cache handling.
bool prepareOption(InitContext &, O &)
bool appendOutput(std::vector< OutputSpec > &, T &, uint32_t)
Outputs handling.
bool finalizeOutput(ProcessingContext &, T &)
bool postRunOutput(EndOfStreamContext &, T &)
bool createExpressionTrees(std::vector< ExpressionInfo > &, T &)
bool prepareOutput(ProcessingContext &, T &)
bool appendOption(std::vector< ConfigParamSpec > &, O &)
Options handling.
bool updateSliceInfo(T &, ArrowTableSlicingCache &)
bool appendCondition(std::vector< InputSpec > &, C &)
Conditions handling.
bool postRunService(EndOfStreamContext &, T &)
bool addService(std::vector< ServiceSpec > &, T &)
Service handling.
bool updatePlaceholders(InitContext &, T &)
Filter handling.
bool registerCache(T &, std::vector< StringPair > &, std::vector< StringPair > &)
Preslice handling.
bool newDataframeCondition(InputRecord &, C &)
void updateFilterInfo(ExpressionInfo &info, std::shared_ptr< arrow::Table > &table)
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
void updatePairList(std::vector< StringPair > &list, std::string const &binding, std::string const &key)
auto homogeneous_apply_refs(L l, T &&object)
DataProcessorSpec adaptAnalysisTask(ConfigContext const &ctx, Args &&... args)
@ Me
Only quit this data processor.
std::string cutString(std::string &&str)
Definition ASoA.cxx:180
auto createFieldsFromColumns(framework::pack< C... >)
Definition ASoA.h:406
void missingFilterDeclaration(int hash, int ai)
Definition ASoA.cxx:28
Defining DataPointCompositeObject explicitly as copiable.
gandiva::Selection selection
Definition Expressions.h:66
size_t processHash
Definition Expressions.h:61
std::function< ProcessCallback(InitContext &)> InitCallback
static std::vector< ServiceSpec > defaultServices(std::string extraPlugins="", int numWorkers=0)
Split a string into a vector of strings using : as a separator.
static std::vector< ServiceSpec > arrowServices()
static void updateInputList(std::vector< InputSpec > &list, InputSpec &&input)
Updates list of InputSpecs by merging metadata.
static constexpr int64_t step
static constexpr int64_t begin
std::vector< std::pair< std::string, bool > > map
Struct to differentiate task names from possible task string arguments.
TaskName(std::string name)
static constexpr uint32_t uniqueId()
From https://en.cppreference.com/w/cpp/utility/variant/visit.
static std::shared_ptr< arrow::Table > joinTables(std::vector< std::shared_ptr< arrow::Table > > &&tables)
Definition ASoA.cxx:67
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"