From 43295ad0002a4d6d1878f15192e94731613e054e Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Mon, 19 Oct 2026 16:14:46 -0400 Subject: [PATCH 1/4] perf(coalesce): specialize Boolean filtering --- arrow-select/src/coalesce.rs | 11 +- arrow-select/src/coalesce/boolean.rs | 217 +++++++++++++++++++++++++++ 2 files changed, 226 insertions(+), 2 deletions(-) create mode 100644 arrow-select/src/coalesce/boolean.rs diff --git a/arrow-select/src/coalesce.rs b/arrow-select/src/coalesce.rs index 6f5378a8f5c8..d0638ea53526 100644 --- a/arrow-select/src/coalesce.rs +++ b/arrow-select/src/coalesce.rs @@ -31,18 +31,24 @@ use std::sync::Arc; // Originally From DataFusion's coalesce module: // https://github.com/apache/datafusion/blob/9d2f04996604e709ee440b65f41e7b882f50b788/datafusion/physical-plan/src/coalesce/mod.rs#L26-L25 +mod boolean; mod byte_view; mod fixed_size_binary; mod generic; mod primitive; +use boolean::InProgressBooleanArray; use byte_view::InProgressByteViewArray; use fixed_size_binary::InProgressFixedSizeBinaryArray; use generic::GenericInProgressArray; use primitive::InProgressPrimitiveArray; fn has_sparse_filter_copy(data_type: &DataType) -> bool { - data_type.is_primitive() || matches!(data_type, DataType::Utf8View | DataType::BinaryView) + data_type.is_primitive() + || matches!( + data_type, + DataType::Boolean | DataType::Utf8View | DataType::BinaryView + ) } /// Maximum selected row fraction for the fused sparse-filter copy path. @@ -783,6 +789,7 @@ fn create_in_progress_array(data_type: &DataType, batch_size: usize) -> Box { Box::new(InProgressByteViewArray::::new(batch_size)) } + DataType::Boolean => Box::new(InProgressBooleanArray::new(batch_size)), DataType::FixedSizeBinary(size) => { Box::new(InProgressFixedSizeBinaryArray::new(*size, batch_size)) } @@ -1771,7 +1778,7 @@ mod tests { true, )])); let coalescer = BatchCoalescer::new(boolean, 100); - assert!(coalescer.has_non_specialized_filter_columns); + assert!(!coalescer.has_non_specialized_filter_columns); } #[derive(Debug, Clone, PartialEq)] diff --git a/arrow-select/src/coalesce/boolean.rs b/arrow-select/src/coalesce/boolean.rs new file mode 100644 index 000000000000..7a2cfec85a6b --- /dev/null +++ b/arrow-select/src/coalesce/boolean.rs @@ -0,0 +1,217 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::InProgressArray; +use crate::filter::{FilterPredicate, FilterSelection}; +use arrow_array::cast::AsArray; +use arrow_array::{Array, ArrayRef, BooleanArray}; +use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder}; +use arrow_schema::ArrowError; +use std::sync::Arc; + +#[derive(Debug)] +pub(crate) struct InProgressBooleanArray { + source: Option, + batch_size: usize, + values: Option, + nulls: NullBufferBuilder, +} + +impl InProgressBooleanArray { + pub(crate) fn new(batch_size: usize) -> Self { + Self { + source: None, + batch_size, + values: None, + nulls: NullBufferBuilder::new(batch_size), + } + } + + fn values(&mut self) -> &mut BooleanBufferBuilder { + self.values + .get_or_insert_with(|| BooleanBufferBuilder::new(self.batch_size)) + } + + fn append_array(&mut self, array: &BooleanArray) { + self.values().append_buffer(array.values()); + if let Some(nulls) = array.nulls() { + self.nulls.append_buffer(nulls); + } else { + self.nulls.append_n_non_nulls(array.len()); + } + } +} + +impl InProgressArray for InProgressBooleanArray { + fn set_source(&mut self, source: Option) { + self.source = source; + } + + fn copy_rows(&mut self, offset: usize, len: usize) -> Result<(), ArrowError> { + let source = self.source.as_ref().ok_or_else(|| { + ArrowError::InvalidArgumentError( + "Internal Error: InProgressBooleanArray: source not set".to_string(), + ) + })?; + let array = source.as_boolean().slice(offset, len); + self.append_array(&array); + Ok(()) + } + + fn copy_rows_by_filter_from( + &mut self, + source: ArrayRef, + filter: &FilterPredicate, + ) -> Result<(), ArrowError> { + // Benchmarks show per-bit copying only wins at selectivities at or below 1/128. + if filter.count() > filter.filter_len() / 128 { + let filtered = filter.filter(source.as_ref())?; + self.append_array(filtered.as_boolean()); + return Ok(()); + } + + let source = source.as_boolean(); + let source_values = source.values(); + if self.values.is_none() { + self.values = Some(BooleanBufferBuilder::new(self.batch_size)); + } + let values = self.values.as_mut().unwrap(); + let output_nulls = &mut self.nulls; + let source_nulls = source.nulls().filter(|nulls| nulls.null_count() != 0); + + match filter.selection() { + FilterSelection::None => {} + FilterSelection::All { len } => { + values.append_packed_range( + source_values.offset()..source_values.offset() + len, + source_values.values(), + ); + if let Some(nulls) = source_nulls { + output_nulls.append_buffer(&nulls.slice(0, len)); + } else { + output_nulls.append_n_non_nulls(len); + } + } + FilterSelection::Slices(slices) => { + if let Some(nulls) = source_nulls { + slices.for_each(|(start, end)| { + values.append_packed_range( + start + source_values.offset()..end + source_values.offset(), + source_values.values(), + ); + output_nulls.append_buffer(&nulls.slice(start, end - start)); + }); + } else { + slices.for_each(|(start, end)| { + values.append_packed_range( + start + source_values.offset()..end + source_values.offset(), + source_values.values(), + ); + }); + output_nulls.append_n_non_nulls(filter.count()); + } + } + FilterSelection::Indices(indices) => { + if let Some(nulls) = source_nulls { + indices.for_each(|index| { + values.append(source_values.value(index)); + output_nulls.append(nulls.is_valid(index)); + }); + } else { + indices.for_each(|index| values.append(source_values.value(index))); + output_nulls.append_n_non_nulls(filter.count()); + } + } + } + Ok(()) + } + + fn finish(&mut self) -> Result { + let values = self + .values + .take() + .map(|mut values| values.finish()) + .unwrap_or_else(|| BooleanBuffer::new_unset(0)); + let nulls = self.nulls.finish(); + self.nulls = NullBufferBuilder::new(self.batch_size); + Ok(Arc::new(BooleanArray::new(values, nulls))) + } + + fn size(&self) -> usize { + self.source + .as_ref() + .map_or(0, |source| source.get_array_memory_size()) + + self + .values + .as_ref() + .map_or(0, |values| values.capacity() / 8) + + self.nulls.allocated_size() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::filter::FilterBuilder; + + #[test] + fn copies_slices_and_sparse_filters() { + let source = Arc::new(BooleanArray::from(vec![ + Some(true), + None, + Some(false), + Some(true), + Some(false), + ])) as ArrayRef; + let mut in_progress = InProgressBooleanArray::new(8); + in_progress.set_source(Some(Arc::clone(&source))); + in_progress.copy_rows(1, 2).unwrap(); + in_progress.set_source(None); + + let filter = BooleanArray::from(vec![true, false, false, true, false]); + let predicate = FilterBuilder::new(&filter).build(); + in_progress + .copy_rows_by_filter_from(source, &predicate) + .unwrap(); + + let output = in_progress.finish().unwrap(); + let expected = BooleanArray::from(vec![None, Some(false), Some(true), Some(true)]); + assert_eq!(output.as_boolean(), &expected); + } + + #[test] + fn finish_resets_buffers() { + let mut in_progress = InProgressBooleanArray::new(8); + assert_eq!(in_progress.finish().unwrap().len(), 0); + + let source = Arc::new(BooleanArray::from(vec![ + Some(true), + None, + Some(false), + None, + ])); + in_progress.set_source(Some(Arc::clone(&source) as ArrayRef)); + in_progress.copy_rows(0, 2).unwrap(); + let first = in_progress.finish().unwrap(); + assert_eq!(first.as_boolean(), &source.slice(0, 2)); + + in_progress.copy_rows(2, 2).unwrap(); + let second = in_progress.finish().unwrap(); + assert_eq!(second.as_boolean(), &source.slice(2, 2)); + assert_eq!(in_progress.finish().unwrap().len(), 0); + } +} From a6c1938487515e23cef7183149b2d6ba46ad087a Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Wed, 21 Oct 2026 14:48:58 -0400 Subject: [PATCH 2/4] refactor(coalesce): simplify sparse Boolean filtering --- arrow-select/src/coalesce/boolean.rs | 173 ++++++++++++++++++--------- 1 file changed, 118 insertions(+), 55 deletions(-) diff --git a/arrow-select/src/coalesce/boolean.rs b/arrow-select/src/coalesce/boolean.rs index 7a2cfec85a6b..32c572743bc0 100644 --- a/arrow-select/src/coalesce/boolean.rs +++ b/arrow-select/src/coalesce/boolean.rs @@ -23,6 +23,8 @@ use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder}; use arrow_schema::ArrowError; use std::sync::Arc; +const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR: usize = 128; + #[derive(Debug)] pub(crate) struct InProgressBooleanArray { source: Option, @@ -78,63 +80,33 @@ impl InProgressArray for InProgressBooleanArray { filter: &FilterPredicate, ) -> Result<(), ArrowError> { // Benchmarks show per-bit copying only wins at selectivities at or below 1/128. - if filter.count() > filter.filter_len() / 128 { - let filtered = filter.filter(source.as_ref())?; - self.append_array(filtered.as_boolean()); - return Ok(()); - } + let indices = match filter.selection() { + FilterSelection::Indices(indices) + if filter.count() + <= filter.filter_len() / BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR => + { + indices + } + _ => { + let filtered = filter.filter(source.as_ref())?; + self.append_array(filtered.as_boolean()); + return Ok(()); + } + }; let source = source.as_boolean(); let source_values = source.values(); - if self.values.is_none() { - self.values = Some(BooleanBufferBuilder::new(self.batch_size)); - } - let values = self.values.as_mut().unwrap(); - let output_nulls = &mut self.nulls; - let source_nulls = source.nulls().filter(|nulls| nulls.null_count() != 0); - - match filter.selection() { - FilterSelection::None => {} - FilterSelection::All { len } => { - values.append_packed_range( - source_values.offset()..source_values.offset() + len, - source_values.values(), - ); - if let Some(nulls) = source_nulls { - output_nulls.append_buffer(&nulls.slice(0, len)); - } else { - output_nulls.append_n_non_nulls(len); - } - } - FilterSelection::Slices(slices) => { - if let Some(nulls) = source_nulls { - slices.for_each(|(start, end)| { - values.append_packed_range( - start + source_values.offset()..end + source_values.offset(), - source_values.values(), - ); - output_nulls.append_buffer(&nulls.slice(start, end - start)); - }); - } else { - slices.for_each(|(start, end)| { - values.append_packed_range( - start + source_values.offset()..end + source_values.offset(), - source_values.values(), - ); - }); - output_nulls.append_n_non_nulls(filter.count()); - } - } - FilterSelection::Indices(indices) => { - if let Some(nulls) = source_nulls { - indices.for_each(|index| { - values.append(source_values.value(index)); - output_nulls.append(nulls.is_valid(index)); - }); - } else { - indices.for_each(|index| values.append(source_values.value(index))); - output_nulls.append_n_non_nulls(filter.count()); - } + let values = self + .values + .get_or_insert_with(|| BooleanBufferBuilder::new(self.batch_size)); + match source.nulls().filter(|nulls| nulls.null_count() != 0) { + Some(nulls) => indices.for_each(|index| { + values.append(source_values.value(index)); + self.nulls.append(nulls.is_valid(index)); + }), + None => { + indices.for_each(|index| values.append(source_values.value(index))); + self.nulls.append_n_non_nulls(filter.count()); } } Ok(()) @@ -166,7 +138,7 @@ impl InProgressArray for InProgressBooleanArray { #[cfg(test)] mod tests { use super::*; - use crate::filter::FilterBuilder; + use crate::filter::{FilterBuilder, FilterIterator}; #[test] fn copies_slices_and_sparse_filters() { @@ -193,6 +165,97 @@ mod tests { assert_eq!(output.as_boolean(), &expected); } + const SPARSE_FILTER_LEN: usize = BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR * 2; + const SOURCE_OFFSET: usize = 3; + const PREFIX_LEN: usize = 3; + + fn sparse_filter_builder() -> FilterBuilder { + let filter = + BooleanArray::from_iter((0..SPARSE_FILTER_LEN).map(|idx| Some(matches!(idx, 5 | 132)))); + FilterBuilder::new(&filter) + } + + fn non_nullable_source() -> BooleanArray { + BooleanArray::from( + (0..SOURCE_OFFSET + SPARSE_FILTER_LEN) + .map(|idx| idx % 3 == 0) + .collect::>(), + ) + } + + fn nullable_source() -> BooleanArray { + BooleanArray::from_iter((0..SOURCE_OFFSET + SPARSE_FILTER_LEN).map(|idx| { + if idx % 8 == 0 { + None + } else { + Some(idx % 3 == 0) + } + })) + } + + fn assert_sparse_copy_matches_filter(source: BooleanArray, predicate: &FilterPredicate) { + assert_eq!( + predicate.count(), + SPARSE_FILTER_LEN / BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR + ); + + let source = Arc::new(source.slice(SOURCE_OFFSET, SPARSE_FILTER_LEN)) as ArrayRef; + let prefix = source.slice(0, PREFIX_LEN); + let filtered = predicate.filter(source.as_ref()).unwrap(); + let expected = crate::concat::concat(&[prefix.as_ref(), filtered.as_ref()]).unwrap(); + + let mut in_progress = InProgressBooleanArray::new(8); + in_progress.set_source(Some(Arc::clone(&source))); + in_progress.copy_rows(0, PREFIX_LEN).unwrap(); + in_progress.set_source(None); + in_progress + .copy_rows_by_filter_from(source, predicate) + .unwrap(); + + let output = in_progress.finish().unwrap(); + assert_eq!(output.as_boolean(), expected.as_boolean()); + } + + #[test] + fn copies_unoptimized_sparse_indices_without_nulls() { + let predicate = sparse_filter_builder().build(); + assert!(matches!( + predicate.selection(), + FilterSelection::Indices(FilterIterator::Lazy(_)) + )); + assert_sparse_copy_matches_filter(non_nullable_source(), &predicate); + } + + #[test] + fn copies_unoptimized_sparse_indices_with_nulls() { + let predicate = sparse_filter_builder().build(); + assert!(matches!( + predicate.selection(), + FilterSelection::Indices(FilterIterator::Lazy(_)) + )); + assert_sparse_copy_matches_filter(nullable_source(), &predicate); + } + + #[test] + fn copies_optimized_sparse_indices_without_nulls() { + let predicate = sparse_filter_builder().optimize().build(); + assert!(matches!( + predicate.selection(), + FilterSelection::Indices(FilterIterator::Materialized(_)) + )); + assert_sparse_copy_matches_filter(non_nullable_source(), &predicate); + } + + #[test] + fn copies_optimized_sparse_indices_with_nulls() { + let predicate = sparse_filter_builder().optimize().build(); + assert!(matches!( + predicate.selection(), + FilterSelection::Indices(FilterIterator::Materialized(_)) + )); + assert_sparse_copy_matches_filter(nullable_source(), &predicate); + } + #[test] fn finish_resets_buffers() { let mut in_progress = InProgressBooleanArray::new(8); From 4c23338354a4735af9add56879c01e820532aabb Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Wed, 21 Oct 2026 15:39:49 -0400 Subject: [PATCH 3/4] perf(coalesce): tune Boolean filter cutoff by nullability --- arrow-select/src/coalesce/boolean.rs | 64 ++++++++++++++++++++-------- 1 file changed, 46 insertions(+), 18 deletions(-) diff --git a/arrow-select/src/coalesce/boolean.rs b/arrow-select/src/coalesce/boolean.rs index 32c572743bc0..457f9ea65148 100644 --- a/arrow-select/src/coalesce/boolean.rs +++ b/arrow-select/src/coalesce/boolean.rs @@ -23,7 +23,11 @@ use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder}; use arrow_schema::ArrowError; use std::sync::Arc; -const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR: usize = 128; +/// With nulls, per-bit copying wins at 0.5% selectivity; use a 1/128 cutoff. +const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS: usize = 128; + +/// Without nulls, per-bit copying wins at 0.1% but loses at 0.5%; 1/256 splits them. +const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS: usize = 256; #[derive(Debug)] pub(crate) struct InProgressBooleanArray { @@ -79,11 +83,14 @@ impl InProgressArray for InProgressBooleanArray { source: ArrayRef, filter: &FilterPredicate, ) -> Result<(), ArrowError> { - // Benchmarks show per-bit copying only wins at selectivities at or below 1/128. + let max_selectivity_denominator = if source.null_count() == 0 { + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS + } else { + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS + }; let indices = match filter.selection() { FilterSelection::Indices(indices) - if filter.count() - <= filter.filter_len() / BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR => + if filter.count() <= filter.filter_len() / max_selectivity_denominator => { indices } @@ -165,13 +172,14 @@ mod tests { assert_eq!(output.as_boolean(), &expected); } - const SPARSE_FILTER_LEN: usize = BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR * 2; + const SPARSE_FILTER_LEN: usize = BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS * 2; const SOURCE_OFFSET: usize = 3; const PREFIX_LEN: usize = 3; - fn sparse_filter_builder() -> FilterBuilder { - let filter = - BooleanArray::from_iter((0..SPARSE_FILTER_LEN).map(|idx| Some(matches!(idx, 5 | 132)))); + fn sparse_filter_builder(selected_indices: &[usize]) -> FilterBuilder { + let filter = BooleanArray::from_iter( + (0..SPARSE_FILTER_LEN).map(|idx| Some(selected_indices.contains(&idx))), + ); FilterBuilder::new(&filter) } @@ -193,10 +201,14 @@ mod tests { })) } - fn assert_sparse_copy_matches_filter(source: BooleanArray, predicate: &FilterPredicate) { + fn assert_sparse_copy_matches_filter( + source: BooleanArray, + predicate: &FilterPredicate, + max_selectivity_denominator: usize, + ) { assert_eq!( predicate.count(), - SPARSE_FILTER_LEN / BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR + SPARSE_FILTER_LEN / max_selectivity_denominator ); let source = Arc::new(source.slice(SOURCE_OFFSET, SPARSE_FILTER_LEN)) as ArrayRef; @@ -218,42 +230,58 @@ mod tests { #[test] fn copies_unoptimized_sparse_indices_without_nulls() { - let predicate = sparse_filter_builder().build(); + let predicate = sparse_filter_builder(&[5]).build(); assert!(matches!( predicate.selection(), FilterSelection::Indices(FilterIterator::Lazy(_)) )); - assert_sparse_copy_matches_filter(non_nullable_source(), &predicate); + assert_sparse_copy_matches_filter( + non_nullable_source(), + &predicate, + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS, + ); } #[test] fn copies_unoptimized_sparse_indices_with_nulls() { - let predicate = sparse_filter_builder().build(); + let predicate = sparse_filter_builder(&[5, 132]).build(); assert!(matches!( predicate.selection(), FilterSelection::Indices(FilterIterator::Lazy(_)) )); - assert_sparse_copy_matches_filter(nullable_source(), &predicate); + assert_sparse_copy_matches_filter( + nullable_source(), + &predicate, + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS, + ); } #[test] fn copies_optimized_sparse_indices_without_nulls() { - let predicate = sparse_filter_builder().optimize().build(); + let predicate = sparse_filter_builder(&[5]).optimize().build(); assert!(matches!( predicate.selection(), FilterSelection::Indices(FilterIterator::Materialized(_)) )); - assert_sparse_copy_matches_filter(non_nullable_source(), &predicate); + assert_sparse_copy_matches_filter( + non_nullable_source(), + &predicate, + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS, + ); } #[test] fn copies_optimized_sparse_indices_with_nulls() { - let predicate = sparse_filter_builder().optimize().build(); + let predicate = sparse_filter_builder(&[5, 132]).optimize().build(); assert!(matches!( predicate.selection(), FilterSelection::Indices(FilterIterator::Materialized(_)) )); - assert_sparse_copy_matches_filter(nullable_source(), &predicate); + assert_sparse_copy_matches_filter( + nullable_source(), + &predicate, + BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS, + ); } #[test] From b4ddd370bce73552241fdd18dd421a31c90b6fb3 Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Fri, 23 Oct 2026 13:24:54 -0400 Subject: [PATCH 4/4] refactor(coalesce): use Boolean filter kernel Always use the existing filter kernel and append its output into the Boolean builders. This keeps the final concat avoidance without benchmark-specific sparse cutoffs. --- arrow-select/src/coalesce/boolean.rs | 157 +-------------------------- 1 file changed, 4 insertions(+), 153 deletions(-) diff --git a/arrow-select/src/coalesce/boolean.rs b/arrow-select/src/coalesce/boolean.rs index 457f9ea65148..17c76b2b6196 100644 --- a/arrow-select/src/coalesce/boolean.rs +++ b/arrow-select/src/coalesce/boolean.rs @@ -16,19 +16,13 @@ // under the License. use super::InProgressArray; -use crate::filter::{FilterPredicate, FilterSelection}; +use crate::filter::FilterPredicate; use arrow_array::cast::AsArray; use arrow_array::{Array, ArrayRef, BooleanArray}; use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder, NullBufferBuilder}; use arrow_schema::ArrowError; use std::sync::Arc; -/// With nulls, per-bit copying wins at 0.5% selectivity; use a 1/128 cutoff. -const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS: usize = 128; - -/// Without nulls, per-bit copying wins at 0.1% but loses at 0.5%; 1/256 splits them. -const BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS: usize = 256; - #[derive(Debug)] pub(crate) struct InProgressBooleanArray { source: Option, @@ -83,39 +77,8 @@ impl InProgressArray for InProgressBooleanArray { source: ArrayRef, filter: &FilterPredicate, ) -> Result<(), ArrowError> { - let max_selectivity_denominator = if source.null_count() == 0 { - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS - } else { - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS - }; - let indices = match filter.selection() { - FilterSelection::Indices(indices) - if filter.count() <= filter.filter_len() / max_selectivity_denominator => - { - indices - } - _ => { - let filtered = filter.filter(source.as_ref())?; - self.append_array(filtered.as_boolean()); - return Ok(()); - } - }; - - let source = source.as_boolean(); - let source_values = source.values(); - let values = self - .values - .get_or_insert_with(|| BooleanBufferBuilder::new(self.batch_size)); - match source.nulls().filter(|nulls| nulls.null_count() != 0) { - Some(nulls) => indices.for_each(|index| { - values.append(source_values.value(index)); - self.nulls.append(nulls.is_valid(index)); - }), - None => { - indices.for_each(|index| values.append(source_values.value(index))); - self.nulls.append_n_non_nulls(filter.count()); - } - } + let filtered = filter.filter(source.as_ref())?; + self.append_array(filtered.as_boolean()); Ok(()) } @@ -145,7 +108,7 @@ impl InProgressArray for InProgressBooleanArray { #[cfg(test)] mod tests { use super::*; - use crate::filter::{FilterBuilder, FilterIterator}; + use crate::filter::FilterBuilder; #[test] fn copies_slices_and_sparse_filters() { @@ -172,118 +135,6 @@ mod tests { assert_eq!(output.as_boolean(), &expected); } - const SPARSE_FILTER_LEN: usize = BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS * 2; - const SOURCE_OFFSET: usize = 3; - const PREFIX_LEN: usize = 3; - - fn sparse_filter_builder(selected_indices: &[usize]) -> FilterBuilder { - let filter = BooleanArray::from_iter( - (0..SPARSE_FILTER_LEN).map(|idx| Some(selected_indices.contains(&idx))), - ); - FilterBuilder::new(&filter) - } - - fn non_nullable_source() -> BooleanArray { - BooleanArray::from( - (0..SOURCE_OFFSET + SPARSE_FILTER_LEN) - .map(|idx| idx % 3 == 0) - .collect::>(), - ) - } - - fn nullable_source() -> BooleanArray { - BooleanArray::from_iter((0..SOURCE_OFFSET + SPARSE_FILTER_LEN).map(|idx| { - if idx % 8 == 0 { - None - } else { - Some(idx % 3 == 0) - } - })) - } - - fn assert_sparse_copy_matches_filter( - source: BooleanArray, - predicate: &FilterPredicate, - max_selectivity_denominator: usize, - ) { - assert_eq!( - predicate.count(), - SPARSE_FILTER_LEN / max_selectivity_denominator - ); - - let source = Arc::new(source.slice(SOURCE_OFFSET, SPARSE_FILTER_LEN)) as ArrayRef; - let prefix = source.slice(0, PREFIX_LEN); - let filtered = predicate.filter(source.as_ref()).unwrap(); - let expected = crate::concat::concat(&[prefix.as_ref(), filtered.as_ref()]).unwrap(); - - let mut in_progress = InProgressBooleanArray::new(8); - in_progress.set_source(Some(Arc::clone(&source))); - in_progress.copy_rows(0, PREFIX_LEN).unwrap(); - in_progress.set_source(None); - in_progress - .copy_rows_by_filter_from(source, predicate) - .unwrap(); - - let output = in_progress.finish().unwrap(); - assert_eq!(output.as_boolean(), expected.as_boolean()); - } - - #[test] - fn copies_unoptimized_sparse_indices_without_nulls() { - let predicate = sparse_filter_builder(&[5]).build(); - assert!(matches!( - predicate.selection(), - FilterSelection::Indices(FilterIterator::Lazy(_)) - )); - assert_sparse_copy_matches_filter( - non_nullable_source(), - &predicate, - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS, - ); - } - - #[test] - fn copies_unoptimized_sparse_indices_with_nulls() { - let predicate = sparse_filter_builder(&[5, 132]).build(); - assert!(matches!( - predicate.selection(), - FilterSelection::Indices(FilterIterator::Lazy(_)) - )); - assert_sparse_copy_matches_filter( - nullable_source(), - &predicate, - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS, - ); - } - - #[test] - fn copies_optimized_sparse_indices_without_nulls() { - let predicate = sparse_filter_builder(&[5]).optimize().build(); - assert!(matches!( - predicate.selection(), - FilterSelection::Indices(FilterIterator::Materialized(_)) - )); - assert_sparse_copy_matches_filter( - non_nullable_source(), - &predicate, - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITHOUT_NULLS, - ); - } - - #[test] - fn copies_optimized_sparse_indices_with_nulls() { - let predicate = sparse_filter_builder(&[5, 132]).optimize().build(); - assert!(matches!( - predicate.selection(), - FilterSelection::Indices(FilterIterator::Materialized(_)) - )); - assert_sparse_copy_matches_filter( - nullable_source(), - &predicate, - BOOLEAN_FILTER_COPY_MAX_SELECTIVITY_DENOMINATOR_WITH_NULLS, - ); - } - #[test] fn finish_resets_buffers() { let mut in_progress = InProgressBooleanArray::new(8);