apache / apache/arrow

HashJoin Operation, need help performing concat of record batches

Open
#33,834 2 comments 0 reactions 0 assignees View on GitHub
Component: C++ Type: usage
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

### Describe the usage question you have. Please include as many useful details as possible.

So I implemented a very inefficient operation to con-cat 2 or more record batches that all individually have an array holding their index. I have been reading about hash operations. I need help improving my current implementation and using arrow functions.

```
pd::DataFrame Concatenator::concatenateColumns(
bool intersect,
bool ignore_index,
bool sort)
{
auto newIndexes = mergeIndexes(makeJoinIndexes(objs, AxisType::Columns), intersect);
const size_t numRows = newIndexes->length();

if (sort)
{
auto sort_indices = ReturnOrThrowOnFailure(arrow::compute::SortIndices(
newIndexes,
arrow::compute::SortOptions{}));

newIndexes =
arrow::compute::Take(newIndexes, sort_indices)->make_array();
}

std::vector index_offset;
index_offset.reserve(objs.size());
auto newColumnLength = accumulate(
objs.begin(),
objs.end(),
0UL,
[&index_offset](size_t total, DataFrame const& df)
{
index_offset.push_back(total);
return total + df.num_columns();
});

arrow::FieldVector fieldVector(newColumnLength);
arrow::ArrayDataVector arrayVectors(newColumnLength);

for(size_t i = 0UL; i < objs.size(); i++)
{
const auto& df = objs[i];
auto schema = df.array()->schema();
auto df_index = df.index();
auto fields = schema->fields();
auto offset = index_offset[i];
for (size_t j = 0UL; j < fields.size(); j++)
{
std::shared_ptr const& columnPerDF = fields[j];
auto col_name = columnPerDF->name();
auto array = df.m_array->GetColumnByName(col_name);
auto array_data = array->data();

if (not array->Equals(newIndexes))
{
auto null = arrow::MakeNullScalar(columnPerDF->type());

arrow::ScalarVector scalars(newIndexes->length(), null);

for (int k = 0; k < newIndexes->length(); k++)
{
auto idx = newIndexes->GetScalar(k).MoveValueUnsafe();
auto result = df_index.index(idx);
if (result != -1)
{
scalars[k] = array->GetScalar(result).MoveValueUnsafe();
}
}
ASSIGN_OR_ABORT(
auto builder,
arrow::MakeBuilder(columnPerDF->type()));

ABORT_NOT_OK(builder->AppendScalars(scalars));

ABORT_NOT_OK(builder->FinishInternal(&array_data));
}

auto flat_index = offset + j;
fieldVector[flat_index] = ignore_index ?
arrow::field(
std::to_string(flat_index),
columnPerDF->type()) :
columnPerDF;

arrayVectors[flat_index] = array_data;
}
}

return { arrow::schema(fieldVector),
static_cast(numRows),
arrayVectors,
newIndexes };

}
```
This is fast as expected but it loses information about the index of each batch
```
pd::DataFrame concatColumnsUnsafe(std::vector const& objs)
{
auto df = objs.at(0).array();
auto N = df->num_columns();

for (int i = 1; i < objs.size(); i++)
{
for (auto const& field : objs[i].array()->schema()->fields())
{
auto result =
df->AddColumn(N++, field, objs[i][field->name()].m_array);
if (result.ok())
{
df = result.MoveValueUnsafe();
}
else
{
throw std::runtime_error(result.status().ToString());
}
}
}
return { df, objs.at(0).indexArray() };
}
```

### Component(s)

C++

Contributor guide

Open the contributing guide

Research direction

Start by reviewing Concatenator::concatenateColumns and concatColumnsUnsafe, then trace how each function constructs and returns the index alongside record batches. Compare the existing Arrow operations with the intended concatenation semantics; done means concatenation preserves the index information for every batch while improving the current implementation.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp
Domain
data
Issue type
Refactor
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.