ARROW-13540: [C++] Add order by sink node #10863
Conversation
|
Draft for now since it'll require rebasing on top of #10793/ARROW-13482 |
|
Now reworked to tag batches instead of being a sink node. Next, I can try to implement a simple order-dependent aggregate kernel and see how far that takes us. |
|
Added a hash_arg_min_max kernel that makes use of the OrderByNode's tag. See ARROW-12873. |
|
Rebased on top of the registry PR. |
|
TODO: revert back to making OrderBy a SinkNode and split arg_min_max and related things into a separate JIRA |
|
Converted back to a sink node. |
bkietz
left a comment
There was a problem hiding this comment.
Some minor comments, overall this looks solid
cpp/src/arrow/compute/exec/options.h
Outdated
There was a problem hiding this comment.
I think this comment needs to be refactored
There was a problem hiding this comment.
Nit: using this above please
| Finishes(ResultWith(::testing::ElementsAreArray(expected)))); | |
| Finishes(ResultWith(ElementsAreArray(expected)))); |
There was a problem hiding this comment.
I think it'd be worthwhile to extract TableFromExecBatches into exec/util.h
There was a problem hiding this comment.
Nit: this would be more readable as
| auto maybe_sorted = SortData(); | |
| Status st = DoFinish(); | |
| if (ErrorIfNotOk(st)) { | |
| producer_.Push(std::move(st)); | |
| } | |
| SinkNode::Finish(); | |
| } | |
| Status DoFinish() { | |
| ARROW_ASSIGN_OR_RAISE(auto sorted, SortData()); | |
| //... |
| .AddToPlan(plan.get())); | ||
|
|
||
| ASSERT_THAT(StartAndCollect(plan.get(), sink_gen), | ||
| Finishes(ResultWith(::testing::ElementsAreArray({ExecBatchFromJSON( |
There was a problem hiding this comment.
| Finishes(ResultWith(::testing::ElementsAreArray({ExecBatchFromJSON( | |
| Finishes(ResultWith(ElementsAreArray({ExecBatchFromJSON( |
| #include "arrow/compute/exec/exec_plan.h" | ||
|
|
||
| #include <mutex> | ||
| #include <unordered_map> |
There was a problem hiding this comment.
| #include <unordered_map> |
Adds a sink node that accumulates, sorts, and emits sorted batches. This is a sink node as currently we don't have a good design for handling order-dependent operations in the rest of the pipeline.