mbutrovich commented on code in PR #11918:
URL: https://github.com/apache/arrow-rs/pull/11918#discussion_r4380936559
##########
arrow-select/src/coalesce/boolean.rs:
##########
@@ -156,4 +332,162 @@ mod tests {
assert_eq!(second.as_boolean(), &source.slice(2, 2));
assert_eq!(in_progress.finish().unwrap().len(), 0);
}
+
+ const FILTER_LEN: usize = 269;
Review Comment:
With `FILTER_LEN = 269`, every case has four full chunks plus a remainder,
and the filter always covers the whole source. Should we add cases for a filter
shorter than the source and for a length that's a multiple of 64, where the
remainder is empty? The coalescer allows a filter shorter than the batch (the
guard at line 206 only rejects a longer one), and none of the tests exercise
either path.
##########
arrow-select/src/coalesce/boolean.rs:
##########
@@ -19,10 +19,136 @@ use super::InProgressArray;
use crate::filter::FilterPredicate;
use arrow_array::cast::AsArray;
use arrow_array::{Array, ArrayRef, BooleanArray};
-use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder};
+use arrow_buffer::bit_chunk_iterator::BitChunks;
+use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder,
bit_util};
use arrow_schema::ArrowError;
use std::sync::Arc;
+/// Joins compressed bit fragments into complete output words.
+#[derive(Default)]
+struct WordPacker {
+ /// Bits waiting to be appended.
+ pending_bits: u64,
+ /// Number of populated low bits in `pending_bits`.
+ pending_len: u32,
+}
+
+impl WordPacker {
+ fn push(&mut self, compressed_bits: u64, bit_count: u32) -> Option<u64> {
+ self.pending_bits |= compressed_bits << self.pending_len;
+ let total_len = self.pending_len + bit_count;
+ if total_len < 64 {
+ self.pending_len = total_len;
+ return None;
+ }
+
+ let complete_word = self.pending_bits;
+ // Keep bits that crossed the word boundary. Split the shift to avoid
+ // shifting by 64 when `pending_len == 0`.
+ self.pending_bits = (compressed_bits >> 1) >> (63 - self.pending_len);
+ self.pending_len = total_len - 64;
+ Some(complete_word)
+ }
+
+ fn finish(self) -> Option<(u64, usize)> {
+ (self.pending_len != 0).then_some((self.pending_bits, self.pending_len
as usize))
+ }
+}
Review Comment:
Can we reuse the packer that `filter_bits_compress` already has instead of
adding a second one?
[`Packer`](https://github.com/apache/arrow-rs/blob/69be95c1d925bd876c71c11ec6cb2a646d54a1ad/arrow-select/src/filter.rs#L812-L839)
in `filter.rs` does the same carry, including the same `(bits >> 1) >> (63 -
filled)` shift, and its
[loop](https://github.com/apache/arrow-rs/blob/69be95c1d925bd876c71c11ec6cb2a646d54a1ad/arrow-select/src/filter.rs#L861-L868)
skips zero mask words before reading the values, as this PR does. Both live in
`arrow-select`, so a shared `pub(crate)` type costs no public API.
Its `push` also takes `(values, mask)` and calls `compress` itself. With
that signature, each compress, push, and append sequence in
`copy_filter_words_without_nulls` and `copy_filter_words_with_nulls` becomes
one call, and you can keep the split into two functions. `filter_bits_compress`
is `#[inline(never)]` because of stack spills, so changing `Packer` to return
words would need a `filter_kernels` run too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
