Project
Loading...
Searching...
No Matches
ASoA.cxx
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#include "Framework/ASoA.h"
13#include "ArrowDebugHelpers.h"
15#include <arrow/util/key_value_metadata.h>
16#include <arrow/util/config.h>
17#include <TMemFile.h>
18#include <TClass.h>
19#include <TTree.h>
20#include <TH1.h>
21#include <TError.h>
22
23namespace o2::soa
24{
25void accessingInvalidIndexFor(const char* getter)
26{
27 throw o2::framework::runtime_error_f("Accessing invalid index for %s", getter);
28}
29void dereferenceWithWrongType(const char* getter, const char* target)
30{
31 throw o2::framework::runtime_error_f("Trying to dereference index with a wrong type in %s_as<T> for base target \"%s\". Note that if you have several compatible index targets in your process() signature, the last one will be the one actually bound.", getter, target);
32}
33void missingFilterDeclaration(int hash, int ai)
34{
35 throw o2::framework::runtime_error_f("Null selection for %d (arg %d), missing Filter declaration?", hash, ai);
36}
37
38void getterNotFound(const char* targetColumnLabel)
39{
40 throw o2::framework::runtime_error_f("Getter for \"%s\" not found", targetColumnLabel);
41}
42
44{
45 throw framework::runtime_error("columnLabel: must not be empty");
46}
47
49{
51 rows.resize(sel->GetNumSlots());
52 for (auto i = 0; i < sel->GetNumSlots(); ++i) {
53 rows[i] = sel->GetIndex(i);
54 }
55 return rows;
56}
57
58SelectionVector sliceSelection(std::span<int64_t const> const& mSelectedRows, int64_t nrows, uint64_t offset)
59{
60 auto start = offset;
61 auto end = start + nrows;
62 auto start_iterator = std::lower_bound(mSelectedRows.begin(), mSelectedRows.end(), start);
63 auto stop_iterator = std::lower_bound(start_iterator, mSelectedRows.end(), end);
64 SelectionVector slicedSelection{start_iterator, stop_iterator};
65 std::transform(slicedSelection.begin(), slicedSelection.end(), slicedSelection.begin(),
66 [&start](int64_t idx) {
67 return idx - static_cast<int64_t>(start);
68 });
69 return slicedSelection;
70}
71
72std::shared_ptr<arrow::Table> ArrowHelpers::joinTables(std::vector<std::shared_ptr<arrow::Table>>&& tables, std::span<const char* const> labels)
73{
74 if (tables.size() == 1) {
75 return tables[0];
76 }
77 for (auto i = 0U; i < tables.size() - 1; ++i) {
78 if (tables[i]->num_rows() != tables[i + 1]->num_rows()) {
79 throw o2::framework::runtime_error_f("Tables %s and %s have different sizes (%d vs %d) and cannot be joined!",
80 labels[i], labels[i + 1], tables[i]->num_rows(), tables[i + 1]->num_rows());
81 }
82 }
83 std::vector<std::shared_ptr<arrow::Field>> fields;
84 std::vector<std::shared_ptr<arrow::ChunkedArray>> columns;
85
86 for (auto& t : tables) {
87 auto tf = t->fields();
88 std::copy(tf.begin(), tf.end(), std::back_inserter(fields));
89 }
90
91 auto schema = std::make_shared<arrow::Schema>(fields);
92
93 if (tables[0]->num_rows() != 0) {
94 for (auto& t : tables) {
95 auto tc = t->columns();
96 std::copy(tc.begin(), tc.end(), std::back_inserter(columns));
97 }
98 }
99 return arrow::Table::Make(schema, columns);
100}
101
102std::shared_ptr<arrow::Table> ArrowHelpers::concatTables(std::vector<std::shared_ptr<arrow::Table>>&& tables)
103{
104 if (tables.size() == 1) {
105 return tables[0];
106 }
107 std::vector<std::shared_ptr<arrow::ChunkedArray>> columns;
108 assert(tables.size() > 1);
109 std::vector<std::shared_ptr<arrow::Field>> resultFields = tables[0]->schema()->fields();
110 auto compareFields = [](std::shared_ptr<arrow::Field> const& f1, std::shared_ptr<arrow::Field> const& f2) {
111 // Let's do this with stable sorting.
112 return (!f1->Equals(f2)) && (f1->name() < f2->name());
113 };
114 for (size_t i = 1; i < tables.size(); ++i) {
115 auto& fields = tables[i]->schema()->fields();
116 std::vector<std::shared_ptr<arrow::Field>> intersection;
117
118 std::set_intersection(resultFields.begin(), resultFields.end(),
119 fields.begin(), fields.end(),
120 std::back_inserter(intersection), compareFields);
121 resultFields.swap(intersection);
122 }
123
124 for (auto& field : resultFields) {
125 arrow::ArrayVector chunks;
126 for (auto& table : tables) {
127 auto ci = table->schema()->GetFieldIndex(field->name());
128 if (ci == -1) {
129 throw std::runtime_error("Unable to find field " + field->name());
130 }
131 auto column = table->column(ci);
132 auto otherChunks = column->chunks();
133 chunks.insert(chunks.end(), otherChunks.begin(), otherChunks.end());
134 }
135 columns.push_back(std::make_shared<arrow::ChunkedArray>(chunks));
136 }
137
138 auto result = arrow::Table::Make(std::make_shared<arrow::Schema>(resultFields), columns);
139 return result;
140}
141
142arrow::ChunkedArray* getIndexFromLabel(arrow::Table* table, std::string_view label)
143{
144 auto field = std::find_if(table->schema()->fields().begin(), table->schema()->fields().end(), [&](std::shared_ptr<arrow::Field> const& f) {
145 auto caseInsensitiveCompare = [](const std::string_view& str1, const std::string& str2) {
146 return std::ranges::equal(
147 str1, str2,
148 [](char c1, char c2) {
149 return std::tolower(static_cast<unsigned char>(c1)) ==
150 std::tolower(static_cast<unsigned char>(c2));
151 });
152 };
153
154 return caseInsensitiveCompare(label, f->name());
155 });
156 if (field == table->schema()->fields().end()) {
157 o2::framework::throw_error(o2::framework::runtime_error_f("Unable to find column with label %s.", label));
158 }
159 auto index = std::distance(table->schema()->fields().begin(), field);
160 return table->column(index).get();
161}
162
163void notBoundTable(const char* tableName)
164{
165 throw o2::framework::runtime_error_f("Index pointing to %s is not bound! Did you subscribe to the table?", tableName);
166}
167
168void notFoundColumn(const char* label, const char* key)
169{
170 throw o2::framework::runtime_error_f(R"(Preslice not valid: table "%s" (or join based on it) does not have column "%s")", label, key);
171}
172
173void missingOptionalPreslice(const char* label, const char* key)
174{
175 throw o2::framework::runtime_error_f(R"(Optional Preslice with missing binding used: table "%s" (or join based on it) does not have column "%s")", label, key);
176}
177
178void* extractCCDBPayload(char* payload, size_t size, TClass const* cl, const char* what)
179{
180 Int_t previousErrorLevel = gErrorIgnoreLevel;
181 gErrorIgnoreLevel = kFatal;
182 // does it have a flattened headers map attached in the end?
183 TMemFile file("name", (char*)payload, size, "READ");
184 gErrorIgnoreLevel = previousErrorLevel;
185 if (file.IsZombie()) {
186 return nullptr;
187 }
188
189 if (!cl) {
190 return nullptr;
191 }
192 auto object = file.GetObjectChecked(what, cl);
193 if (!object) {
194 // it could be that object was stored with previous convention
195 // where the classname was taken as key
196 std::string objectName(cl->GetName());
197 objectName.erase(std::find_if(objectName.rbegin(), objectName.rend(), [](unsigned char ch) {
198 return !std::isspace(ch);
199 }).base(),
200 objectName.end());
201 objectName.erase(objectName.begin(), std::find_if(objectName.begin(), objectName.end(), [](unsigned char ch) {
202 return !std::isspace(ch);
203 }));
204
205 object = file.GetObjectChecked(objectName.c_str(), cl);
206 LOG(warn) << "Did not find object under expected name " << what;
207 if (!object) {
208 return nullptr;
209 }
210 LOG(warn) << "Found object under deprecated name " << cl->GetName();
211 }
212 auto result = object;
213 // We need to handle some specific cases as ROOT ties them deeply
214 // to the file they are contained in
215 if (cl->InheritsFrom("TObject")) {
216 // make a clone
217 // detach from the file
218 auto tree = dynamic_cast<TTree*>((TObject*)object);
219 if (tree) {
220 tree->LoadBaskets(0x1L << 32); // make tree memory based
221 tree->SetDirectory(nullptr);
222 result = tree;
223 } else {
224 auto h = dynamic_cast<TH1*>((TObject*)object);
225 if (h) {
226 h->SetDirectory(nullptr);
227 result = h;
228 }
229 }
230 }
231 return result;
232}
233
234} // namespace o2::soa
235
236namespace o2::framework
237{
238std::string cutString(std::string&& str)
239{
240 auto pos = str.find('_');
241 if (pos != std::string::npos) {
242 str.erase(pos);
243 }
244 return str;
245}
246
247std::string strToUpper(std::string&& str)
248{
249 std::transform(str.begin(), str.end(), str.begin(), [](unsigned char c) { return std::toupper(c); });
250 return str;
251}
252
254{
255 return binding == "[MISSING]";
256}
257
259{
260 return bindingKey;
261}
262
267
272
273std::shared_ptr<arrow::Table> PreslicePolicySorted::getSliceFor(int value, std::shared_ptr<arrow::Table> const& input, uint64_t& offset) const
274{
275 auto [offset_, count] = this->sliceInfo.getSliceFor(value);
276 auto output = input->Slice(offset_, count);
277 offset = static_cast<int64_t>(offset_);
278 return output;
279}
280
281std::span<const int64_t> PreslicePolicyGeneral::getSliceFor(int value) const
282{
283 return this->sliceInfo.getSliceFor(value);
284}
285} // namespace o2::framework
int32_t i
void output(const std::map< std::string, ChannelStat > &channels)
Definition rawdump.cxx:197
uint16_t pos
Definition RawData.h:3
uint32_t c
Definition RawData.h:2
StringRef key
Class for time synchronization of RawReader instances.
GLint GLsizei count
Definition glcorearb.h:399
GLuint64EXT * result
Definition glcorearb.h:5662
GLsizeiptr size
Definition glcorearb.h:659
GLuint GLuint end
Definition glcorearb.h:469
GLuint index
Definition glcorearb.h:781
GLdouble f
Definition glcorearb.h:310
GLsizei const GLfloat * value
Definition glcorearb.h:819
GLenum target
Definition glcorearb.h:1641
GLintptr offset
Definition glcorearb.h:660
GLuint GLsizei const GLchar * label
Definition glcorearb.h:2519
GLuint object
Definition glcorearb.h:4041
GLuint start
Definition glcorearb.h:469
std::shared_ptr< gandiva::SelectionVector > Selection
Definition Expressions.h:46
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
RuntimeErrorRef runtime_error(const char *)
void throw_error(RuntimeErrorRef)
std::string strToUpper(std::string &&str)
Definition ASoA.cxx:247
RuntimeErrorRef runtime_error_f(const char *,...)
std::string cutString(std::string &&str)
Definition ASoA.cxx:238
void * extractCCDBPayload(char *payload, size_t size, TClass const *cl, const char *what)
Definition ASoA.cxx:178
SelectionVector selectionToVector(gandiva::Selection const &sel)
Definition ASoA.cxx:48
void notBoundTable(const char *tableName)
Definition ASoA.cxx:163
SelectionVector sliceSelection(std::span< int64_t const > const &mSelectedRows, int64_t nrows, uint64_t offset)
Definition ASoA.cxx:58
std::vector< int64_t > SelectionVector
Definition ASoA.h:415
void missingFilterDeclaration(int hash, int ai)
Definition ASoA.cxx:33
void accessingInvalidIndexFor(const char *getter)
Definition ASoA.cxx:25
void dereferenceWithWrongType(const char *getter, const char *target)
Definition ASoA.cxx:29
void emptyColumnLabel()
Definition ASoA.cxx:43
void getterNotFound(const char *targetColumnLabel)
Definition ASoA.cxx:38
void missingOptionalPreslice(const char *label, const char *key)
Definition ASoA.cxx:173
arrow::ChunkedArray * getIndexFromLabel(arrow::Table *table, std::string_view label)
Definition ASoA.cxx:142
void notFoundColumn(const char *label, const char *key)
Definition ASoA.cxx:168
std::unique_ptr< GPUReconstructionTimeframe > tf
const std::string binding
Definition ASoA.h:1431
Entry const & getBindingKey() const
Definition ASoA.cxx:258
SliceInfoUnsortedPtr sliceInfo
Definition ASoA.h:1448
std::span< const int64_t > getSliceFor(int value) const
Definition ASoA.cxx:281
void updateSliceInfo(SliceInfoUnsortedPtr &&si)
Definition ASoA.cxx:268
void updateSliceInfo(SliceInfoPtr &&si)
Definition ASoA.cxx:263
std::shared_ptr< arrow::Table > getSliceFor(int value, std::shared_ptr< arrow::Table > const &input, uint64_t &offset) const
Definition ASoA.cxx:273
std::pair< int64_t, int64_t > getSliceFor(int value) const
std::span< int64_t const > getSliceFor(int value) const
static std::shared_ptr< arrow::Table > joinTables(std::vector< std::shared_ptr< arrow::Table > > &&tables, std::span< const char *const > labels)
Definition ASoA.cxx:72
static std::shared_ptr< arrow::Table > concatTables(std::vector< std::shared_ptr< arrow::Table > > &&tables)
Definition ASoA.cxx:102
LOG(info)<< "Compressed in "<< sw.CpuTime()<< " s"
std::unique_ptr< TTree > tree((TTree *) flIn.Get(std::string(o2::base::NameConf::CTFTREENAME).c_str()))
std::vector< ReadoutWindowData > rows
const std::string str