Skip to content

Commit 72432db

Browse files
aalkinaalkin
committed
use serialized pointer
1 parent d7b88e5 commit 72432db

3 files changed

Lines changed: 53 additions & 13 deletions

File tree

‎Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx‎

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111

1212
#include "AnalysisCCDBHelpers.h"
1313
#include "CCDBFetcherHelper.h"
14+
#include "Framework/ArrowTypes.h"
1415
#include "Framework/DataProcessingStats.h"
1516
#include "Framework/DeviceSpec.h"
1617
#include "Framework/TimingInfo.h"
@@ -29,6 +30,8 @@
2930
#include <arrow/table.h>
3031
#include <arrow/array.h>
3132
#include <arrow/builder.h>
33+
#include <arrow/io/memory.h>
34+
#include <arrow/ipc/writer.h>
3235
#include <fmt/base.h>
3336
#include <ctime>
3437
#include <memory>
@@ -109,7 +112,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
109112
auto it = ccdbUrls.find(m.name);
110113
fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
111114
auto columnName = m.name.substr(strlen("ccdb:"));
112-
fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, fieldMetadata));
115+
fields.emplace_back(std::make_shared<arrow::Field>(columnName, soa::asArrowDataType<int64_t[2]>(), false, fieldMetadata));
113116
}
114117
schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
115118
}
@@ -122,6 +125,7 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
122125
return adaptStateless([schemas, bindings, helper](InputRecord& inputs, DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo, DataProcessingStats& stats) {
123126
O2_SIGNPOST_ID_GENERATE(sid, ccdb);
124127
O2_SIGNPOST_START(ccdb, sid, "fetchFromAnalysisCCDB", "Fetching CCDB objects for analysis%" PRIu64, (uint64_t)timingInfo.timeslice);
128+
auto pool = arrow::MemoryPool::CreateDefault();
125129
for (auto& schema : schemas) {
126130
std::vector<CCDBFetcherHelper::FetchOp> ops;
127131
auto inputBinding = *schema->metadata()->Get("sourceTable");
@@ -143,15 +147,23 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
143147
}
144148
int outputRouteIndex = bindings.at(outRouteDesc);
145149
auto& spec = helper->routes[outputRouteIndex].matcher;
146-
std::vector<std::shared_ptr<arrow::BinaryViewBuilder>> builders;
147-
for (auto const& _ : schema->fields()) {
148-
builders.emplace_back(std::make_shared<arrow::BinaryViewBuilder>());
150+
std::vector<std::shared_ptr<arrow::FixedSizeListBuilder>> builders;
151+
builders.resize(schema->fields().size());
152+
153+
for (auto i = 0U; i < schema->fields().size(); ++i) {
154+
auto valueBuilder = std::make_shared<arrow::Int64Builder>();
155+
builders[i] = std::make_shared<arrow::FixedSizeListBuilder>(pool.get(), valueBuilder, 2);
149156
}
150157

151158
auto reserveSize = timestampColumn->length();
159+
O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
160+
"* reserving for size: %lld (has: %lld)",
161+
reserveSize, builders[0]->capacity());
152162
arrow::Status status;
153163
for (auto i = 0U; i < builders.size(); ++i) {
154-
status &= builders[i]->Reserve(reserveSize);
164+
if (builders[i]->capacity() < reserveSize) {
165+
status &= builders[i]->Reserve(reserveSize - builders[i]->capacity());
166+
}
155167
}
156168

157169
for (auto ci = 0; ci < timestampColumn->num_chunks(); ++ci) {
@@ -181,11 +193,16 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
181193
LOGP(fatal, "Not enough responses (expected {}, found {})", builders.size(), responses.size());
182194
}
183195
arrow::Status result;
196+
int64_t values[2];
184197
for (size_t bi = 0; bi < responses.size(); bi++) {
185198
auto& builder = builders[bi];
199+
auto* value_builder = static_cast<arrow::Int64Builder*>(builder->value_builder());
186200
auto& response = responses[bi];
187-
char const* address = reinterpret_cast<char const*>(response.id.value);
188-
result &= builder->Append(std::string_view(address, response.size));
201+
values[0] = response.id.value;
202+
values[1] = response.size;
203+
result &= builder->Append();
204+
result &= value_builder->AppendValues(&values[0], 2, nullptr);
205+
LOGP(info, "P: {}; S: {}", values[0], values[1]);
189206
}
190207
if (!result.ok()) {
191208
LOGP(fatal, "Error adding results from CCDB");
@@ -198,6 +215,17 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
198215
arrays.push_back(*builder->Finish());
199216
}
200217
auto outTable = arrow::Table::Make(schema, arrays);
218+
219+
auto mock = std::make_shared<arrow::io::MockOutputStream>();
220+
int64_t expectedSize = 0;
221+
auto mockWriter = arrow::ipc::MakeStreamWriter(mock.get(), outTable->schema());
222+
arrow::Status outStatus = mockWriter.ValueOrDie()->WriteTable(*(outTable.get()));
223+
224+
expectedSize = mock->Tell().ValueOrDie();
225+
assert(outTable->num_rows() == reserveSize);
226+
O2_SIGNPOST_EVENT_EMIT_INFO(ccdb, sid, "fetchFromAnalysisCCDB",
227+
"* sending a table of size: %lld",
228+
expectedSize);
201229
auto concrete = DataSpecUtils::asConcreteDataMatcher(spec);
202230
allocator.adopt(Output{concrete.origin, concrete.description, concrete.subSpec}, outTable);
203231
}

‎Framework/Core/include/Framework/ASoA.h‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2455,15 +2455,15 @@ consteval static std::string_view namespace_prefix()
24552455
[[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }
24562456

24572457
#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_) \
2458-
struct _Name_ : o2::soa::Column<std::span<std::byte>, _Name_> { \
2458+
struct _Name_ : o2::soa::Column<int64_t[2], _Name_> { \
24592459
static constexpr const char* mLabel = _Label_; \
24602460
static constexpr const char* query = _CCDBQuery_; \
24612461
static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_}); \
2462-
using base = o2::soa::Column<std::span<std::byte>, _Name_>; \
2463-
using type = std::span<std::byte>; \
2462+
using base = o2::soa::Column<int64_t[2], _Name_>; \
2463+
using type = int64_t[2]; \
24642464
using column_t = _Name_; \
24652465
_Name_(arrow::ChunkedArray const* column) \
2466-
: o2::soa::Column<std::span<std::byte>, _Name_>(o2::soa::ColumnIterator<std::span<std::byte>>(column)) \
2466+
: o2::soa::Column<int64_t[2], _Name_>(o2::soa::ColumnIterator<type>(column)) \
24672467
{ \
24682468
} \
24692469
\
@@ -2473,13 +2473,15 @@ consteval static std::string_view namespace_prefix()
24732473
\
24742474
decltype(auto) _Getter_() const \
24752475
{ \
2476+
auto a = *mColumnIterator; \
2477+
LOGP(info, "P: {}; S: {}", a[0], a[1]); \
2478+
auto span = std::span<std::byte>{reinterpret_cast<std::byte*>(a[0]), static_cast<size_t>(a[1])}; \
24762479
if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) { \
2477-
return *mColumnIterator; \
2480+
return span; \
24782481
} else { \
24792482
static std::byte* payload = nullptr; \
24802483
static _ConcreteType_* deserialised = nullptr; \
24812484
static TClass* c = TClass::GetClass(#_ConcreteType_); \
2482-
auto span = *mColumnIterator; \
24832485
if (payload != (std::byte*)span.data()) { \
24842486
payload = (std::byte*)span.data(); \
24852487
delete deserialised; \

‎Framework/Core/include/Framework/ArrowTypes.h‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,16 @@ struct arrow_array_for<int8_t[N]> {
9393
using type = arrow::FixedSizeListArray;
9494
using value_type = int8_t;
9595
};
96+
template <int N>
97+
struct arrow_array_for<int64_t[N]> {
98+
using type = arrow::FixedSizeListArray;
99+
using value_type = int64_t;
100+
};
101+
template <int N>
102+
struct arrow_array_for<uint64_t[N]> {
103+
using type = arrow::FixedSizeListArray;
104+
using value_type = uint64_t;
105+
};
96106

97107
#define ARROW_VECTOR_FOR(_type_) \
98108
template <> \

0 commit comments

Comments
 (0)