Skip to content

DPL Analysis: avoid memory churn related to slices #15447

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ktf wants to merge 1 commit into from
+38 −5
Closed

DPL Analysis: avoid memory churn related to slices #15447

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 38 additions & 5 deletions Framework/Core/include/Framework/ASoA.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2088,6 +2088,7 @@ class Table
Table(std::shared_ptr<arrow::Table> table, uint64_t offset = 0)
: mTable(table),
mOffset(offset),
mSize{table->num_rows()},
mEnd{table->num_rows()}
{
if (mTable->num_rows() == 0) {
Expand All @@ -2105,6 +2106,30 @@ class Table
}
}

/// Construct a view over a sub-range of an existing table without
/// calling arrow::Table::Slice (zero heap allocations).
Table(std::shared_ptr<arrow::Table> table, uint64_t offset, uint64_t count)
: mTable(table),
mOffset(0),
mSize{static_cast<int64_t>(count)},
mEnd{static_cast<int64_t>(offset + count)}
{
if (count == 0) {
for (size_t ci = 0; ci < framework::pack_size(columns_t{}); ++ci) {
mColumnChunks[ci] = nullptr;
}
mBegin = mEnd;
} else {
auto lookups = [this]<typename... C>(framework::pack<C...>) -> std::array<arrow::ChunkedArray*, framework::pack_size(columns_t{})> { return {lookupColumn<C>()...}; }(columns_t{});
for (size_t ci = 0; ci < framework::pack_size(columns_t{}); ++ci) {
mColumnChunks[ci] = lookups[ci];
}
mBegin = unfiltered_iterator{mColumnChunks, {static_cast<int64_t>(offset + count), 0}};
mBegin.setCursor(offset);
mBegin.bindInternalIndices(this);
}
}

Table(std::vector<std::shared_ptr<arrow::Table>>&& tables, uint64_t offset = 0)
requires(ref.origin_hash != "CONC"_h)
: Table(ArrowHelpers::joinTables(std::move(tables), std::span{originalLabels}), offset)
Expand Down Expand Up @@ -2201,7 +2226,7 @@ class Table
/// Size of the table, in rows.
[[nodiscard]] int64_t size() const
{
return mTable->num_rows();
return mSize;
}

[[nodiscard]] int64_t tableSize() const
Expand Down Expand Up @@ -2287,12 +2312,12 @@ class Table

auto rawSlice(uint64_t start, uint64_t end) const
{
return self_t{mTable->Slice(start, end - start + 1), start};
return self_t{mTable, start, end - start + 1};
}

auto emptySlice() const
{
return self_t{mTable->Slice(0, 0), 0};
return self_t{mTable, 0, 0};
}

private:
Expand All @@ -2308,6 +2333,7 @@ class Table
}
std::shared_ptr<arrow::Table> mTable = nullptr;
uint64_t mOffset = 0;
int64_t mSize = 0;
// Cached pointers to the ChunkedArray associated to a column
arrow::ChunkedArray* mColumnChunks[framework::pack_size(columns_t{})];
RowViewSentinel mEnd;
Expand Down Expand Up @@ -3470,6 +3496,13 @@ struct Join : Table<o2::aod::Hash<"JOIN"_h>, o2::aod::Hash<"JOIN/0"_h>, o2::aod:
bindInternalIndicesTo(this);
}
}
Join(std::shared_ptr<arrow::Table> table, uint64_t offset, uint64_t count)
: base{table, offset, count}
{
if (count != 0) {
bindInternalIndicesTo(this);
}
}
using base::bindExternalIndices;
using base::bindInternalIndicesTo;
static constexpr const uint32_t binding_origin = base::binding_origin;
Expand Down Expand Up @@ -3539,12 +3572,12 @@ struct Join : Table<o2::aod::Hash<"JOIN"_h>, o2::aod::Hash<"JOIN/0"_h>, o2::aod:

auto rawSlice(uint64_t start, uint64_t end) const
{
return self_t{{this->asArrowTable()->Slice(start, end - start + 1)}, start};
return self_t{this->asArrowTable(), start, end - start + 1};
}

auto emptySlice() const
{
return self_t{{this->asArrowTable()->Slice(0, 0)}, 0};
return self_t{this->asArrowTable(), 0, 0};
}

template <typename T>
Expand Down