GH-47376: [C++][Compute] Support selective execution for kernels#47377
GH-47376: [C++][Compute] Support selective execution for kernels#47377zanmato1984 wants to merge 8 commits into
Conversation
|
Thanks for opening a pull request! If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project. Then could you also rename the pull request title in the following format? or See also: |
zanmato1984
changed the title
|
|
|
|
) ### Rationale for this change In order to support special form (#47374), the kernels have to respect the selection vector. Currently none of the kernels does. And it's almost impossible for us to make all existing kernels to respect the selection vector at once (and we probably never will). Thus we need an incremental way to add selection-vector-aware kernels on demand, meanwhile accommodate legacy (selection-vector-non-aware) kernels to be executed "selection-vector-aware"-ly in a general manner - the idea is to first "gather" selected rows from the batch into a new batch, evaluate the expression on the new batch, then "scatter" the result rows into the positions where they belong in the original batch. This makes the `take` and `scatter` functions dependencies of the exec facilities, which is in compute core (libarrow). And `take` is already in compute core. Now we need to move `scatter`. I'm implementing the selective execution of kernels in #47377, including invoking `take` and `scatter` as explained above. And I have to write tests of that in `exec_test.cc` which is deliberately declared to be NOT depending on libarrow_compute. ### What changes are included in this PR? Move scatter compute function into compute core. ### Are these changes tested? Yes. Manually tested. ### Are there any user-facing changes? None. * GitHub Issue: #47375 Authored-by: Rossi Sun <zanmato1984@gmail.com> Signed-off-by: Rossi Sun <zanmato1984@gmail.com>
zanmato1984
force-pushed
the
fix/gh-47376
branch
2 times, most recently
from
0cce8c3 to
99a3217
Compare
zanmato1984
force-pushed
the
fix/gh-47376
branch
2 times, most recently
from
99098d2 to
c005bbb
Compare
zanmato1984
force-pushed
the
fix/gh-47376
branch
7 times, most recently
from
c502acc to
469ead1
Compare
zanmato1984
force-pushed
the
fix/gh-47376
branch
from
469ead1 to
9a1c49e
Compare
|
Attaching benchmark results. The benchmark employs a trivial kernel that does nothing but spins specified number of times to simulate CPU intensity, and the number of rows of the batch is 4k. Baseline: Regular kernel, no selection vector. click to expandSparse: Selective kernel, with a selection vector of selectivity from 0 to 100%. click to expandDense: Regular kernel enclosed by gather/scatter, with a selection vector of selectivity from 0 to 100%. click to expand |
|
Some interesting comparisons to note:
|
|
Hi @pitrou @bkietz @westonpace @felipecrv , I know this is a big one, but I do hope some of you can help to review this PR - this is the most critical prerequisite for the if_else special form. Appreciated! |
zanmato1984
force-pushed
the
fix/gh-47376
branch
4 times, most recently
from
966d999 to
9072a34
Compare
This reverts commit db970f1e89722dcf65b09553c51ed1b1a4ede3ce.
| std::vector<Datum> values(batch.num_values()); | ||
| for (int i = 0; i < batch.num_values(); ++i) { | ||
| if (batch[i].is_scalar()) { | ||
| // XXX: Skip gather for scalars since it is not currently supported by Take. |
There was a problem hiding this comment.
Technically it's not necessary. But the drawback is we lose the ability to uniformly call Take on any Datum - have to make sure it's not scalar and go through a special path, like here, for scalar.
I think maybe we can simply return the scalar as is for Take (to allow the uniform invoking on arbitrary Datum). Or we insist that taking scalar makes no sense and we do special checks everywhere.
| /// handling (intersect validity bitmaps of inputs). | ||
| /// \brief Add a kernel with given input/output types and exec API, no selective exec | ||
| /// API, no required state initialization, preallocation for fixed-width types, and | ||
| /// default null handling (intersect validity bitmaps of inputs). |
There was a problem hiding this comment.
I think this is being as clear as the rest of the style ("no required state" etc.)
| using ArrayKernelSelectiveExec = Status (*)(KernelContext*, const ExecSpan&, | ||
| const SelectionVectorSpan&, ExecResult*); |
There was a problem hiding this comment.
Good suggestion. Shall we do that in follow-up PRs?
| while (indices_begin + num_indices < indices_end && | ||
| *(indices_begin + num_indices) < chunk_row_id_end) { | ||
| ++num_indices; | ||
| } |
There was a problem hiding this comment.
Good catch. Replaced with a log(N) complexity std::lower_bound().
| return kernel_->selective_exec(kernel_ctx_, input, *selection, out); | ||
| } | ||
| return kernel_->exec(kernel_ctx_, input, out); | ||
| } |
There was a problem hiding this comment.
This pre-condition that non-null selection implies non-null
selective_execis very specific.
Sorry I don't get it. The two callsites both have the possibility that selection is non-null, and we need to make sure that selective_exec is also non-null. In other word, if we inline it, the code would be exactly the same in these two places.
Or are you suggesting something performance-wise?
| void SetSlice(int64_t offset, int64_t length, int32_t index_back_shift = 0); | ||
|
|
||
| int32_t operator[](int64_t i) const { | ||
| return indices_[i + offset_] - index_back_shift_; |
There was a problem hiding this comment.
My concern is that exposing the index_back_shift would be too verbose and error-prone. Better use some encapsulation to hide it. Maybe let the span accept a lambda, within which we can write more compiler-friendly code meanwhile keep the index_back_shift hidden?
| VisitSelectionVectorSpanInline(const SelectionVectorSpan& selection, | ||
| OnSelectionFn&& on_selection) { | ||
| for (int64_t i = 0; i < selection.length(); ++i) { | ||
| RETURN_NOT_OK(on_selection(selection[i])); |
There was a problem hiding this comment.
Sorry, I don't get it. Could you elaborate a bit?
|
Kindly ping @felipecrv @pitrou . Thanks. |
|
Hi @pitrou @felipecrv , could you please kindly help? Thanks. |
github-actions
Bot
added
awaiting changes
|
Kindly ping @felipecrv @pitrou . Thanks. |
|
Please kindly help @felipecrv @pitrou . Thanks! |
|
Really really need you guys' help :) @pitrou @felipecrv |
|
Ping @pitrou @felipecrv . I really want to move this forward, and a subsequent feature enabling special forms is awaiting. Please help, thanks! |
|
Hey @zanmato1984, sorry for not responding earlier. Reviewing this is a non-trivial task, especially as I'm worried about the API and efficiency implications of this. I'm also worried about what I perceive as additional complexity in the execution layer, which is already overly complicated with lots of weird cases (I'm afraid bugs and hidden inefficiencies will hide there). I'm also skeptical that a selection vector that is purely a vector of selection indices is a good solution performance-wise. In many cases you might have long spans of contiguous selected values, but with this solution we lose any opportunity of efficient batch execution (not to mention potential SIMD vectorization). This might be especially painful with chunked inputs (does this PR work with chunked arrays at all?). This is also my experience following the take/filter implementation work that we have been doing heroic efforts to try and improve the indirect indexing story while IMHO we should really be talking in terms of contiguous spans, or more precisely a mixed representation (spans/individual indices). All in all, I'm afraid this is committing to a lot of choices that both add complexity while painting us in a corner performance-wise. What do you think? You probably have thought about all this already. (also for the record my current professional situation is that I'm available for smaller maintenance and PR reviews but it's difficult to commit non-client time for such large enhancements) |
|
@pitrou Thanks a lot for taking the time and for the candid feedback - totally understood on the time constraints. I agree the execution layer is already overly complicated. In this PR I tried to keep the added complexity moderate and localized: when no selection vector is present, the existing execution path is unchanged (aside from a couple of null checks in the iterator). The selection-aware path is only activated when an ExecBatch carries a On performance: the selection path is only intended to be exercised from the upcoming special-form work (#47374), as a narrow and semantically explicit entry point. The primary goal there is semantic correctness under vectorized execution (e.g. guarding against errors like division-by-zero in rows where the condition is false), and I think paying some overhead is acceptable for that. I posted benchmark results earlier (#47377 (comment)); in summary, sparse execution wins strongly at low selectivity, while the worst regressions (up to ~4x) are from the generic dense fallback when a kernel doesn’t provide Chunked inputs: yes, this PR works with chunked arrays. Both the chunked + selection cases are covered by unit tests ( Finally, I hear you on representation: an index-only SelectionVector loses contiguous-span information and may limit batch/SIMD opportunities. I haven’t fully designed a mixed representation yet, but I agree it may be the right long-term direction. If you have a preferred API shape (hybrid runs+indices, a generalized “selection” object, or a different exec signature), I’d really appreciate guidance - I’d rather adjust before we cement the API. |
|
First I forgot another problem: the PR currently uses 32-bit indices, but we really want 64-bit indices, right? (perhaps UInt64 to match what sort_indices outputs, though that's unnecessary).
Separately from this PR, can we think about ways to make it simpler? Perhaps there are internal execution "options" that aren't really useful.
Yes, but we would like it to be more generally useful for execution engines, right?
The regression might be much worse on chunked inputs?
I'm not sure what it should look like, and we can probably add some complexity piecewise if we agree the API remains experimental. Ideally I'd like something that can be used internally for take/filter as well. A conceptual sketch could look like: struct ContiguousSpan {
int64_t start_offset;
int64_t length;
};
struct FilteredSpan {
int64_t start_offset;
int64_t length;
/* followed by a filter bitmap with `length` bits */
};
struct DiscreteSpan {
int64_t length;
/* followed by `length` 64-bit indices */
};
using SelectionSpan = std::variant<ContiguousSpan, FilteredSpan, DiscreteSpan>;(but SelectionSpan would actually be encoded using some bit-twiddling and a selection vector would be a Buffer containing a number of SelectionSpans) That's of course quite a bit of work and DiscreteSpan might be the only implemented variant at the start. |
- Deduplicate ApplySelectionMask for ArrayData/ArraySpan\n- Fix OUTPUT_NOT_NULL full-span validity init\n- Add unit test for OUTPUT_NOT_NULL selective exec
- Centralize SelectedCount/empty-span checks\n- Reduce scattered selection if/else in ScalarExecutor
apache#47378) ### Rationale for this change In order to support special form (apache#47374), the kernels have to respect the selection vector. Currently none of the kernels does. And it's almost impossible for us to make all existing kernels to respect the selection vector at once (and we probably never will). Thus we need an incremental way to add selection-vector-aware kernels on demand, meanwhile accommodate legacy (selection-vector-non-aware) kernels to be executed "selection-vector-aware"-ly in a general manner - the idea is to first "gather" selected rows from the batch into a new batch, evaluate the expression on the new batch, then "scatter" the result rows into the positions where they belong in the original batch. This makes the `take` and `scatter` functions dependencies of the exec facilities, which is in compute core (libarrow). And `take` is already in compute core. Now we need to move `scatter`. I'm implementing the selective execution of kernels in apache#47377, including invoking `take` and `scatter` as explained above. And I have to write tests of that in `exec_test.cc` which is deliberately declared to be NOT depending on libarrow_compute. ### What changes are included in this PR? Move scatter compute function into compute core. ### Are these changes tested? Yes. Manually tested. ### Are there any user-facing changes? None. * GitHub Issue: apache#47375 Authored-by: Rossi Sun <zanmato1984@gmail.com> Signed-off-by: Rossi Sun <zanmato1984@gmail.com>
3 participants
Rationale for this change
In order to support special form (#47374), being able to "selective"-ly execute the kernel becomes a prerequisite. As mentioned in #47374, we need an incremental way to add selective kernels on demand, meanwhile accommodate arbitrary legacy kernels to be executed selectively in a general manner.
What changes are included in this PR?
ArrayKernelSelectiveExec(KernelContext*, const ExecSpan&, const SelectionVectorSpan&, ExecResult*)in the kernel. This is the entry for selectively executing the kernel on a batch with a given selection vector. The kernel author can provide a dedicated implementation for such kernel API so the kernel can be executed "sparse"-ly - only the rows indicated by the selection vector will be processed. Otherwise the selective execution will fall back to a general "dense" way - gather the selected rows into a new contiguous (dense) batch, execute the kernel using the non-selective exec API, then scatter the result back to the original row positions.ScalarExecutorwith dense execution ability.Are these changes tested?
Tested and benchmarked.
Are there any user-facing changes?
None.