Skip to content
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

Fix Partition<Filtered>, add separate sliceByCached for joins, filtered and nested filtered #8626

Merged
merged 4 commits into from
Apr 22, 2022
Merged

Fix Partition<Filtered>, add separate sliceByCached for joins, filtered and nested filtered #8626

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
370bb59
Introduce separate sliceByCached for joins, filtered, partitions. Fix
saganatt Apr 20, 2022
12f594a
Simplify logics
saganatt Apr 21, 2022
f101659
Nitpick
saganatt Apr 21, 2022
c266312
Lambda capture
saganatt Apr 21, 2022
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
85 changes: 79 additions & 6 deletions Framework/Core/include/Framework/ASoA.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2133,6 +2133,20 @@ struct Join : JoinBase<Ts...> {
using const_iterator = iterator;
using filtered_iterator = typename table_t::template RowViewFiltered<Join<Ts...>, Ts...>;
using filtered_const_iterator = filtered_iterator;

auto sliceByCached(framework::expressions::BindingNode const& node, int value)
{
uint64_t offset = 0;
std::shared_ptr<arrow::Table> result = nullptr;
auto status = this->getSliceFor(value, node.name.c_str(), result, offset);
if (status.ok()) {
auto t = Join<Ts...>({result}, offset);
this->copyIndexBindings(t);
return t;
}
o2::framework::throw_error(o2::framework::runtime_error("Failed to slice table"));
O2_BUILTIN_UNREACHABLE();
saganatt marked this conversation as resolved.
Show resolved Hide resolved
}
};

template <typename... Ts>
Expand Down Expand Up @@ -2321,6 +2335,11 @@ class FilteredBase : public T
std::shared_ptr<arrow::Table> result = nullptr;
auto status = ((table_t*)this)->getSliceFor(value, node.name.c_str(), result, offset);
if (status.ok()) {
if (offset >= this->tableSize()) {
self_t fresult{{result}, SelectionVector{}, 0}; // empty slice
this->copyIndexBindings(fresult);
return fresult;
}
auto start = offset;
auto end = start + result->num_rows();
auto start_iterator = std::lower_bound(mSelectedRows.begin(), mSelectedRows.end(), start);
Expand Down Expand Up @@ -2441,6 +2460,8 @@ class Filtered : public FilteredBase<T>
{
public:
using self_t = Filtered<T>;
using table_t = typename FilteredBase<T>::table_t;

Filtered(std::vector<std::shared_ptr<arrow::Table>>&& tables, gandiva::Selection const& selection, uint64_t offset = 0)
: FilteredBase<T>(std::move(tables), selection, offset) {}

Expand Down Expand Up @@ -2521,7 +2542,35 @@ class Filtered : public FilteredBase<T>
{
return operator*=(other.getSelectedRows());
}
using FilteredBase<T>::sliceByCached;

auto sliceByCached(framework::expressions::BindingNode const& node, int value)
{
uint64_t offset = 0;
std::shared_ptr<arrow::Table> result = nullptr;
auto status = ((table_t*)this)->getSliceFor(value, node.name.c_str(), result, offset);
if (status.ok()) {
if (offset >= this->tableSize()) {
self_t fresult{{result}, SelectionVector{}, 0}; // empty slice
this->copyIndexBindings(fresult);
return fresult;
}
auto start = offset;
auto end = start + result->num_rows();
auto start_iterator = std::lower_bound(this->getSelectedRows().begin(), this->getSelectedRows().end(), start);
auto stop_iterator = std::lower_bound(start_iterator, this->getSelectedRows().end(), end);
SelectionVector slicedSelection{start_iterator, stop_iterator};
std::transform(slicedSelection.begin(), slicedSelection.end(), slicedSelection.begin(),
[&](int64_t idx) {
return idx - static_cast<int64_t>(start);
});
auto slicedSize = slicedSelection.size();
self_t fresult{{result}, std::move(slicedSelection), start};
this->copyIndexBindings(fresult);
return fresult;
}
o2::framework::throw_error(o2::framework::runtime_error("Failed to slice table"));
O2_BUILTIN_UNREACHABLE();
}
};

template <typename T>
Expand All @@ -2532,23 +2581,23 @@ class Filtered<Filtered<T>> : public FilteredBase<typename T::table_t>
using table_t = typename FilteredBase<typename T::table_t>::table_t;

Filtered(std::vector<Filtered<T>>&& tables, gandiva::Selection const& selection, uint64_t offset = 0)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(std::move(tables))), selection, offset)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(tables)), selection, offset)
{
for (auto& table : tables) {
*this *= table;
}
}

Filtered(std::vector<Filtered<T>>&& tables, SelectionVector&& selection, uint64_t offset = 0)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(std::move(tables))), std::forward<SelectionVector>(selection), offset)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(tables)), std::forward<SelectionVector>(selection), offset)
{
for (auto& table : tables) {
*this *= table;
}
}

Filtered(std::vector<Filtered<T>>&& tables, gsl::span<int64_t const> const& selection, uint64_t offset = 0)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(std::move(tables))), selection, offset)
: FilteredBase<typename T::table_t>(std::move(extractTablesFromFiltered(tables)), selection, offset)
{
for (auto& table : tables) {
*this *= table;
Expand Down Expand Up @@ -2627,10 +2676,34 @@ class Filtered<Filtered<T>> : public FilteredBase<typename T::table_t>
return operator*=(other.getSelectedRows());
}

using FilteredBase<typename T::table_t>::sliceByCached;
auto sliceByCached(framework::expressions::BindingNode const& node, int value)
{
uint64_t offset = 0;
std::shared_ptr<arrow::Table> result = nullptr;
auto status = ((table_t*)this)->getSliceFor(value, node.name.c_str(), result, offset);
if (status.ok()) {
auto start = offset;
auto end = start + result->num_rows();
auto start_iterator = std::lower_bound(this->getSelectedRows().begin(), this->getSelectedRows().end(), start);
ktf marked this conversation as resolved.
Show resolved Hide resolved
auto stop_iterator = std::lower_bound(start_iterator, this->getSelectedRows().end(), end);
SelectionVector slicedSelection{start_iterator, stop_iterator};
std::transform(slicedSelection.begin(), slicedSelection.end(), slicedSelection.begin(),
[&](int64_t idx) {
return idx - static_cast<int64_t>(start);
});
SelectionVector copy = slicedSelection;
Filtered<T> filteredTable{{result}, std::move(slicedSelection), start};
std::vector<Filtered<T>> filtered{filteredTable};
self_t fresult{std::move(filtered), std::move(copy), start};
this->copyIndexBindings(fresult);
return fresult;
}
o2::framework::throw_error(o2::framework::runtime_error("Failed to slice table"));
O2_BUILTIN_UNREACHABLE();
}

private:
std::vector<std::shared_ptr<arrow::Table>> extractTablesFromFiltered(std::vector<Filtered<T>>&& tables)
std::vector<std::shared_ptr<arrow::Table>> extractTablesFromFiltered(std::vector<Filtered<T>>& tables)
{
std::vector<std::shared_ptr<arrow::Table>> outTables;
for (auto& table : tables) {
Expand Down