Add coalesce kernel andBatchCoalescer for statefully combining selected b…atches: (#7597)

# Which issue does this PR close?

- Part of #6692
- Part of #7589


# Rationale for this change

The pattern of combining multiple small RecordBatches to form one larger
one for
subsequent processing is common in query engines like DataFusion which
filter or
partition incoming Arrays. Current best practice is to use the `filter`
or `take` kernels and then
`concat` kernels as explained in
- #6692

This pattern also appears in my attempt to improve parquet filter
performance (to cache the result
of applying a filter rather than re-decoding the results). See
- apache/datafusion#3463
- #7513

The current pattern is non optimal as it requires:
1. At least 2x peak memory (holding the input and output of `concat`)
2. 2 copies of the data (to create the output of `filter` and then
create the output of `concat`)

The theory is that with sufficient optimization we can reduce the peak
memory
requirements and (possibly) make it faster as well.

However, to add a benchmark for this filter+concat, I basically had
nothing to
benchmark. Specifically, there needed to be an API to call.

- Note I also made a PR to DataFusion showing this API can be used and
it is not slower: apache/datafusion#16249


# What changes are included in this PR?

I ported the code from DataFusion downstream upstream into arrow-rs so
1. We can use it in the parquet reader
2. We can benchmark and optimize it appropriately

1. Add `BatchCoalescer` to `arrow-select`, and tests
2. Update documentation
2. Add examples
3. Add a `pub` export in `arrow`
4. Add Benchmark

# Are there any user-facing changes?

This is a new API. 

I next plan to make an benchmark for this particular

Author: alamb

arrow-select/src/coalesce.rs

@@ -156,10 +156,16 @@ pub fn prep_null_mask_filter(filter: &BooleanArray) -> BooleanArray {
     BooleanArray::new(mask, None)
 }
 
-/// Returns a filtered `values` [Array] where the corresponding elements of
+/// Returns a filtered `values` [`Array`] where the corresponding elements of
 /// `predicate` are `true`.
 ///
-/// See also [`FilterBuilder`] for more control over the filtering process.
+/// # See also
+/// * [`FilterBuilder`] for more control over the filtering process.
+/// * [`filter_record_batch`] to filter a [`RecordBatch`]
+/// * [`BatchCoalescer`]: to filter multiple [`RecordBatch`] and coalesce
+///   the results into a single array.
+///
+/// [`BatchCoalescer`]: crate::coalesce::BatchCoalescer
 ///
 /// # Example
 /// ```rust

arrow-select/src/filter.rs

@@ -24,6 +24,7 @@
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(missing_docs)]
 
+pub mod coalesce;
 pub mod concat;
 mod dictionary;
 pub mod filter;

arrow-select/src/lib.rs

@@ -64,6 +64,12 @@ use num::{One, Zero};
 ///
 /// When `options` is not set to check bounds, taking indexes after `len` will panic.
 ///
+/// # See also
+/// * [`BatchCoalescer`]: to filter multiple [`RecordBatch`] and coalesce
+///   the results into a single array.
+///
+/// [`BatchCoalescer`]: crate::coalesce::BatchCoalescer
+///
 /// # Examples
 /// ```
 /// # use arrow_array::{StringArray, UInt32Array, cast::AsArray};

arrow-select/src/take.rs

@@ -161,6 +161,12 @@ name = "filter_kernels"
 harness = false
 required-features = ["test_utils"]
 
+[[bench]]
+name = "coalesce_kernels"
+harness = false
+required-features = ["test_utils"]
+
+
 [[bench]]
 name = "take_kernels"
 harness = false