arrow::compute `hash_*` functions matching input types

[jrecuerda]

I am not able to use the hash_* compute functions. I have tried both in python and directly in C++ but I am always getting the following error:

Function hash_sum has no kernel matching input types (array[int64], array[string])

I have tried multiple combinations of input types: float-int64, float-string, double-string, double-int64...

These are two simplified examples of what I'm trying:

Python

>>> import pyarrow.compute as pc
>>> pc.sum([1,2,3])
<pyarrow.Int64Scalar: 6>
>>> pc.hash_sum([1,2,3], ["a", "b", "a"])
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/vscode/.local/lib/python3.6/site-packages/pyarrow/compute.py", line 191, in wrapper
    return func.call(args, options, memory_pool)
  File "pyarrow/_compute.pyx", line 327, in pyarrow._compute.Function.call
  File "pyarrow/error.pxi", line 143, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 120, in pyarrow.lib.check_status
pyarrow.lib.ArrowNotImplementedError: Function hash_sum has no kernel matching input types (array[int64], array[string])

C++

        arrow::DoubleBuilder builder;
        builder.Append(1);
        builder.Append(2);
        builder.Append(3);
        auto array = builder.Finish();

        arrow::StringBuilder builderStr;
        builderStr.Append("a");
        builderStr.Append("b");
        builderStr.Append("a");
        auto arrayKey = builderStr.Finish();

        std::vector<std::shared_ptr<arrow::Field>> schema_vector = {
            arrow::field("id", arrow::float64()), arrow::field("cat", arrow::utf8())};

        auto schema = std::make_shared<arrow::Schema>(schema_vector);
        auto table = arrow::Table::Make(schema, {*array, *arrayKey});
        auto res = arrow::compute::CallFunction(std::string("hash_count"), {table->GetColumnByName("id"), table->GetColumnByName("cat")});

What am I missing?

[AlenkaF]

In the python binding (pyarrow) the hash_* functions are not implemented to be called directly.

If you use a development version of pyarrow you could do something like this:

import pyarrow as pa
import pyarrow.compute as pc

table = pa.table([[1, 2, 3],["a", "b", "a"]], names=["values", "keys"])

table.group_by("keys").aggregate([("values","hash_sum")])

and you would get:

pyarrow.Table
values_sum: int64
keys: string
----
values_sum: [[4,2]]
keys: [["a","b"]]

[jrecuerda]

Thank you. Is it planned to be released soon? Why the C++ code gives me the same error?

[AlenkaF]

Releases are planned quarterly https://arrow.apache.org/release/.
As for the C++ I think it should work on grouped data also.

[jrecuerda]

As I mentioned the C++ code in the initial post gives me the error below. Should I open an issue in JIRA?
no kernel matching input types (array[int64], array[string])

[AlenkaF]

What I mean is that it should work in C++ also if not called directly. Here I am not of much help but would maybe try with something like:

arrow::compute::CountOptions all(arrow::compute::CountOptions::ALL);
auto res =
    arrow::compute::internal::GroupBy({array}, {arrayKey}, {{"hash_count", &all}}, false);

[westonpace]

The hash_* functions all take, as the last argument, a uint32 array of group ids which explains the error. However, even if you were to correct this you would get the error: Direct execution of HASH_AGGREGATE functions.

At the time (6.0.1) this was prevented. I think because we didn't want to confuse people that expected something more like a "group by" operation that both computes the group ids and performs the aggregate. The calculation of group ids is not currently exposed because it is stateful and we haven't exposed any stateful kernels.

So, at the moment, I think you may be out of luck with 6.0.1. I believe the only way to utilize these would be to use the query plan directly and that hasn't been documented yet (other than dplyr). Work is being done to document the query plans in C++ and to expose the query plans in python via ibis and there has been some discussion on exposing the hash kernels and grouping kernels directly as it could be useful and simple for "dataset-in-memory" operations. And, of course, there is the approach that Alenka shared. So I think you will have several options once 7.0.0 releases.

If you're interested in an undocumented C++ approach I could share a snippet on how you could use a query plan to accomplish what you want. Although I would have to know a bit more what you were trying to accomplish. Were you intending to group by the string column and return the sums of the int64 column?

[jrecuerda]

Thank you for the details. I just want to compute the average of a table column grouped by another column as it is shown in the docs.

[XinyuZeng]

Exactly fall into the same issue here. I think the cpp doc mentioned by @jrecuerda above is a little confusing because the "Invoking functions" section teaches us the way to use CallFunction and "Grouped Aggregations" section lists out the names and input args. Then a natural way for me to do group by is arrow::compute::CallFunction but it does not work

[westonpace]

Sorry for the delay but here is an update on a few things.

We are working on documenting the C++ exec plans API here: #12033

The exec plans API for 6.0.1 (and 7.0.0) is still pretty rough around the edges. I've jotted down some potential improvements here: https://issues.apache.org/jira/browse/ARROW-15257

As mentioned earlier, a simpler API for aggregations is being added to pyarrow as part of 7.0.0 which will be releasing fairly soon (within the next month).

As promised I have sketched up a minimal (though not very simple) C++ example of averaging a table grouped by a column: https://gist.github.com/westonpace/e555a3b1c269c31de7176d34f47a2fb0

[edponce]

@XinyuZeng Thanks for informing us parts of the documentation that can be misleading. We are working on improving documentation and examples for the different functionalities of Arrow. At the moment, the description of "Grouped Aggregations" states that these type of functions are not directly invocable. Nevertheless, I understand this is easy to miss when browsing through a list of tables, and moreover, when these are the only functions with such a constraint. I created ARROW-15269 to clarify this concern.

[jrecuerda]

Thank you very much for the detailed response. I managed to perform the operation over a table by using the following snippet:

void compute_histogram(const arrow::Table &table, int bin_size, int max)
{
    auto plan = cp::ExecPlan::Make().ValueOrDie();
    auto reader = std::make_shared<arrow::TableBatchReader>(table);

    // auto context = arrow::io::IOContext();
    // auto batch_gen = cp::MakeReaderGenerator(std::move(reader), context.executor()).ValueOrDie();
    auto thread_pool = arrow::internal::GetCpuThreadPool();
    auto batch_gen = cp::MakeReaderGenerator(std::move(reader), thread_pool).ValueOrDie();

    cp::CountOptions options(cp::CountOptions::ONLY_VALID);
    arrow::AsyncGenerator<util::optional<cp::ExecBatch>> sink_gen;
    cp::Declaration::Sequence({
                                  {"source", cp::SourceNodeOptions{table.schema(), batch_gen}},
                                  {"project",
                                   cp::ProjectNodeOptions{{cp::field_ref("duration_histogram.duration"),
                                                           cp::call("cast",
                                                                    {cp::call("divide",
                                                                              {cp::field_ref("duration_histogram.duration"),
                                                                               cp::literal(bin_size)})},
                                                                    std::make_shared<cp::CastOptions>(cp::CastOptions::Unsafe(arrow::int64())))},
                                                          {"duration", "duration_bin"}}},

                                  {"aggregate",
                                   cp::AggregateNodeOptions{
                                       /*aggregates=*/{{"hash_count", &options}},
                                       /*targets=*/{"duration"},
                                       /*names=*/{"result"},
                                       /*keys=*/{"duration_bin"}}},
                                  {"sink", cp::SinkNodeOptions{&sink_gen}},
                              })
        .AddToPlan(plan.get())
        .ValueOrDie();

    if (auto status = plan->Validate(); !status.ok())
    {
        throw std::runtime_error("Plan validation error: " + status.ToString());
    }

    if (auto status = plan->StartProducing(); !status.ok())
    {
        throw std::runtime_error("Production error: " + status.ToString());
    }

    auto outputSchema = arrow::schema({arrow::field("duration", arrow::float64()),
                                       arrow::field("duration_bin", arrow::int64())});

    auto collected_fut = arrow::CollectAsyncGenerator(sink_gen);
    auto res = arrow::AllComplete({plan->finished(), arrow::Future<>(collected_fut)})
                   .Then([collected_fut, outputSchema]() -> arrow::Result<std::vector<std::shared_ptr<arrow::RecordBatch>>>
                         {
        auto collected = collected_fut.result().ValueOrDie();
        // std::vector<cp::ExecBatch> out;
        std::vector<std::shared_ptr<arrow::RecordBatch>> out;
        out.reserve(collected.size());
        std::transform(std::make_move_iterator(collected.begin()),
                       std::make_move_iterator(collected.end()), std::back_inserter(out),
                       [outputSchema](util::optional<cp::ExecBatch> batch)
                    //    { return std::move(*batch); });
                       { return batch->ToRecordBatch(outputSchema).ValueOrDie(); });
        return out; });

    auto outputTable = arrow::Table::FromRecordBatches(outputSchema, res.result().ValueOrDie()).ValueOrDie();
    std::cout << outputTable->ToString() << "\n";
}

[rohanjain101]

@westonpace is there an updated snippet for calling hash aggregate functions in arrow 14? I am trying to call the hash aggregate functions directly on an arrow table as well from c++ and facing similar issues as original post about direct execution of hash aggregate functions not allowed. I tried to use the snippet you shared (https://gist.github.com/westonpace/e555a3b1c269c31de7176d34f47a2fb0) but I am not able to compile it with arrow 14. Is there another way to achieve this with later versions of arrow?