From 94f0bcb02f28374929547bc7c7073a25bae252cb Mon Sep 17 00:00:00 2001 From: Nikita Matskevich Date: Sat, 24 Oct 2026 13:59:20 +0200 Subject: [PATCH 1/2] feat(parquet): allow disabling Bloom filter folding --- parquet/src/arrow/arrow_writer/byte_array.rs | 6 +- parquet/src/arrow/arrow_writer/mod.rs | 78 ++++++++++++++++++++ parquet/src/column/writer/encoder.rs | 14 ++-- parquet/src/file/properties.rs | 56 ++++++++++++-- 4 files changed, 138 insertions(+), 16 deletions(-) diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs b/parquet/src/arrow/arrow_writer/byte_array.rs index 70687874b82f..98b0a3f9d744 100644 --- a/parquet/src/arrow/arrow_writer/byte_array.rs +++ b/parquet/src/arrow/arrow_writer/byte_array.rs @@ -429,7 +429,7 @@ pub struct ByteArrayEncoder { min_value: Option, max_value: Option, bloom_filter: Option, - bloom_filter_target_fpp: f64, + bloom_filter_target_fpp: Option, geo_stats_accumulator: Option>, is_decimal: bool, } @@ -439,7 +439,9 @@ impl ColumnValueEncoder for ByteArrayEncoder { type Values = dyn Array; fn flush_bloom_filter(&mut self) -> Option { let mut sbbf = self.bloom_filter.take()?; - sbbf.fold_to_target_fpp(self.bloom_filter_target_fpp); + if let Some(target_fpp) = self.bloom_filter_target_fpp { + sbbf.fold_to_target_fpp(target_fpp); + } Some(sbbf) } diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs index 8ce51de52d07..2d164d5f0ec7 100644 --- a/parquet/src/arrow/arrow_writer/mod.rs +++ b/parquet/src/arrow/arrow_writer/mod.rs @@ -3440,6 +3440,84 @@ mod tests { ); } + #[test] + fn bloom_filter_folding_can_be_disabled_per_column() { + use crate::bloom_filter::Sbbf; + use crate::file::properties::BloomFilterProperties; + + for dictionary_enabled in [false, true] { + let arrays: Vec = vec![ + Arc::new(Int32Array::from_iter_values(0..100)), + Arc::new(StringArray::from_iter_values( + (0..100).map(|i| i.to_string()), + )), + ]; + for array in arrays { + let schema = Arc::new(Schema::new(vec![ + Field::new("folded", array.data_type().clone(), false), + Field::new("retained", array.data_type().clone(), false), + ])); + let batch = + RecordBatch::try_new(schema.clone(), vec![array.clone(), array]).unwrap(); + let bloom_props = |folding_enabled| { + BloomFilterProperties::builder() + .with_max_ndv(1000) + .with_fpp(0.01) + .with_folding_enabled(folding_enabled) + .build() + }; + let props = WriterProperties::builder() + .set_dictionary_enabled(dictionary_enabled) + .set_bloom_filter_properties(bloom_props(true)) + .set_column_bloom_filter_properties( + ColumnPath::from("retained"), + bloom_props(false), + ) + .build(); + let mut buf = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut buf, schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let options = ReadOptionsBuilder::new() + .with_reader_properties( + ReaderProperties::builder() + .set_read_bloom_filter(true) + .build(), + ) + .build(); + let reader = + SerializedFileReader::new_with_options(Bytes::from(buf), options).unwrap(); + let row_group = reader.get_row_group(0).unwrap(); + let folded = row_group.get_column_bloom_filter(0).unwrap(); + let retained = row_group.get_column_bloom_filter(1).unwrap(); + let mut expected = Sbbf::new_with_ndv_fpp(1000, 0.01).unwrap(); + for i in 0..100 { + match batch.column(0).data_type() { + DataType::Int32 => { + assert!(folded.check(&i)); + assert!(retained.check(&i)); + expected.insert(&i); + } + DataType::Utf8 => { + let value = i.to_string(); + assert!(folded.check(value.as_str())); + assert!(retained.check(value.as_str())); + expected.insert(value.as_str()); + } + _ => unreachable!(), + } + } + let serialize = |sbbf: &Sbbf| { + let mut bytes = Vec::new(); + sbbf.write(&mut bytes).unwrap(); + bytes + }; + assert_eq!(serialize(retained), serialize(&expected)); + assert!(serialize(folded).len() < serialize(retained).len()); + } + } + } + fn write_with_bloom_filter(array: ArrayRef, dictionary_page_size_limit: usize) -> Bytes { let schema = Arc::new(Schema::new(vec![Field::new( "col", diff --git a/parquet/src/column/writer/encoder.rs b/parquet/src/column/writer/encoder.rs index c0d10aeb95d0..7b4887310b5a 100644 --- a/parquet/src/column/writer/encoder.rs +++ b/parquet/src/column/writer/encoder.rs @@ -205,7 +205,7 @@ pub struct ColumnValueEncoderImpl { max_value: Option, nan_count: Option, bloom_filter: Option, - bloom_filter_target_fpp: f64, + bloom_filter_target_fpp: Option, variable_length_bytes: Option, geo_stats_accumulator: Option>, } @@ -261,7 +261,9 @@ impl ColumnValueEncoder for ColumnValueEncoderImpl { fn flush_bloom_filter(&mut self) -> Option { let mut sbbf = self.bloom_filter.take()?; - sbbf.fold_to_target_fpp(self.bloom_filter_target_fpp); + if let Some(target_fpp) = self.bloom_filter_target_fpp { + sbbf.fold_to_target_fpp(target_fpp); + } Some(sbbf) } @@ -455,16 +457,16 @@ impl ColumnValueEncoder for ColumnValueEncoderImpl { } /// Creates a bloom filter sized for the column's configured NDV, returning the filter -/// and the target FPP for folding. +/// and the target FPP for folding, or `None` when folding is disabled. pub(crate) fn create_bloom_filter( column_props: &ResolvedColumnProperties, -) -> Result<(Option, f64)> { +) -> Result<(Option, Option)> { match column_props.bloom_filter_properties.as_ref() { Some(bf_props) => Ok(( Some(Sbbf::new_with_ndv_fpp(bf_props.ndv(), bf_props.fpp())?), - bf_props.fpp(), + bf_props.folding_enabled().then_some(bf_props.fpp()), )), - None => Ok((None, 0.0)), + None => Ok((None, None)), } } diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs index 6fc7b822add8..cd59493c21c7 100644 --- a/parquet/src/file/properties.rs +++ b/parquet/src/file/properties.rs @@ -1548,7 +1548,8 @@ impl Default for EnabledStatistics { /// The bloom filter is initially sized for `ndv` distinct values at the given `fpp`, then /// automatically folded down after all values are inserted to achieve optimal size while /// maintaining the target `fpp`. See [`Sbbf::fold_to_target_fpp`] for details on the -/// folding algorithm. +/// folding algorithm. Disable folding with +/// [`BloomFilterPropertiesBuilder::with_folding_enabled`] to retain the initial allocation. /// /// # Example /// @@ -1578,6 +1579,7 @@ impl Default for EnabledStatistics { pub struct BloomFilterProperties { fpp: f64, ndv: u64, + folding_enabled: bool, } impl Default for BloomFilterProperties { @@ -1585,6 +1587,7 @@ impl Default for BloomFilterProperties { BloomFilterProperties { fpp: DEFAULT_BLOOM_FILTER_FPP, ndv: DEFAULT_BLOOM_FILTER_NDV, + folding_enabled: true, } } } @@ -1605,11 +1608,19 @@ impl BloomFilterProperties { /// e.g. 0.1, 0.05, or 0.001 is recommended. /// /// This value also serves as the target FPP for bloom filter folding: after all values - /// are inserted, the filter is folded down to the smallest size that still meets this FPP. + /// are inserted, the filter is folded down to the smallest size that still meets this FPP + /// when [`Self::folding_enabled`] is true. pub fn fpp(&self) -> f64 { self.fpp } + /// Whether the writer folds sparse filters after insertion. Defaults to `true`. + /// + /// Set with [`BloomFilterPropertiesBuilder::with_folding_enabled`]. + pub fn folding_enabled(&self) -> bool { + self.folding_enabled + } + /// Maximum expected number of distinct values. Defaults to [`DEFAULT_BLOOM_FILTER_NDV`]. /// /// You should set this value by calling [`WriterPropertiesBuilder::set_bloom_filter_max_ndv`]. @@ -1617,12 +1628,12 @@ impl BloomFilterProperties { /// When not explicitly set via the builder, this defaults to /// [`max_row_group_row_count`](WriterProperties::max_row_group_row_count) (resolved at /// build time). The bloom filter is initially sized for this many distinct values at the - /// given `fpp`, then folded down after insertion to achieve optimal size. A good heuristic + /// given `fpp`, then folded down after insertion if folding is enabled. A good heuristic /// is to set this to the expected number of rows in the row group. If fewer distinct values - /// are actually written, the filter will be automatically compacted via folding. + /// are actually written, the filter will be automatically compacted when folding is enabled. /// - /// Thus the only negative side of overestimating this value is that the bloom filter - /// will use more memory during writing than necessary, but it will not affect the final + /// With folding enabled, overestimating this value makes the bloom filter use more memory + /// during writing than necessary, but it will not affect the final /// bloom filter size on disk. /// /// If you wish to reduce memory usage during writing and are able to make a reasonable estimate @@ -1644,6 +1655,7 @@ impl BloomFilterProperties { pub struct BloomFilterPropertiesBuilder { fpp: Option, ndv: Option, + folding_enabled: Option, } impl BloomFilterPropertiesBuilder { @@ -1685,6 +1697,18 @@ impl BloomFilterPropertiesBuilder { self } + /// Enables or disables folding after values are inserted (defaults to `true`). + /// + /// Disabling folding retains the initial NDV/FPP-sized allocation, matching + /// the non-adaptive parquet-java writer's behavior. This can preserve a lower + /// realized false positive probability for sparse row groups at the cost of + /// a larger serialized filter. It does not change hashing or initial sizing, + /// and does not by itself guarantee byte identity with other writers. + pub fn with_folding_enabled(mut self, enabled: bool) -> Self { + self.folding_enabled = Some(enabled); + self + } + /// Builds [`BloomFilterProperties`]. /// /// @@ -1702,7 +1726,12 @@ impl BloomFilterPropertiesBuilder { let fpp = self.fpp.unwrap_or(DEFAULT_BLOOM_FILTER_FPP); validate_bloom_filter_fpp(fpp).map_err(ParquetError::General)?; let ndv = self.ndv.unwrap_or(DEFAULT_BLOOM_FILTER_NDV); - Ok(BloomFilterProperties { fpp, ndv }) + let folding_enabled = self.folding_enabled.unwrap_or(true); + Ok(BloomFilterProperties { + fpp, + ndv, + folding_enabled, + }) } } @@ -2480,7 +2509,11 @@ mod tests { ); assert_eq!( props.bloom_filter_properties(&ColumnPath::from("col")), - Some(&BloomFilterProperties { fpp: 0.1, ndv: 100 }) + Some(&BloomFilterProperties { + fpp: 0.1, + ndv: 100, + ..Default::default() + }) ); } @@ -2518,6 +2551,7 @@ mod tests { Some(&BloomFilterProperties { fpp: DEFAULT_BLOOM_FILTER_FPP, ndv: DEFAULT_BLOOM_FILTER_NDV, + ..Default::default() }) ); } @@ -2562,6 +2596,7 @@ mod tests { Some(&BloomFilterProperties { fpp: DEFAULT_BLOOM_FILTER_FPP, ndv: 100, + ..Default::default() }) ); assert_eq!( @@ -2572,6 +2607,7 @@ mod tests { Some(&BloomFilterProperties { fpp: 0.1, ndv: DEFAULT_BLOOM_FILTER_NDV, + ..Default::default() }) ); } @@ -2617,6 +2653,7 @@ mod tests { Some(&BloomFilterProperties { fpp: DEFAULT_BLOOM_FILTER_FPP, ndv: 100, + ..Default::default() }) ); assert_eq!( @@ -2624,6 +2661,7 @@ mod tests { Some(&BloomFilterProperties { fpp: DEFAULT_BLOOM_FILTER_FPP, ndv: 200, + ..Default::default() }) ); } @@ -2776,6 +2814,7 @@ mod tests { let props = BloomFilterProperties::builder().build(); assert_eq!(props.fpp, DEFAULT_BLOOM_FILTER_FPP); assert_eq!(props.ndv, DEFAULT_BLOOM_FILTER_NDV); + assert!(props.folding_enabled()); assert_eq!(props, BloomFilterProperties::default()); assert_eq!( BloomFilterPropertiesBuilder::new().build(), @@ -2877,6 +2916,7 @@ mod tests { .build(); let tailored = BloomFilterProperties::builder() .with_fpp(0.02) + .with_folding_enabled(false) .with_max_ndv(1000) .build(); From d7defcd1d769d74e8179ffd2d5d344e5931059e8 Mon Sep 17 00:00:00 2001 From: Nikita Matskevich Date: Sat, 24 Oct 2026 14:21:11 +0200 Subject: [PATCH 2/2] test(parquet): preserve extra pruning when folding is disabled --- parquet/src/arrow/arrow_writer/mod.rs | 29 ++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs index 2d164d5f0ec7..874adfdfc3aa 100644 --- a/parquet/src/arrow/arrow_writer/mod.rs +++ b/parquet/src/arrow/arrow_writer/mod.rs @@ -3445,6 +3445,7 @@ mod tests { use crate::bloom_filter::Sbbf; use crate::file::properties::BloomFilterProperties; + let target_fpp = 0.01; for dictionary_enabled in [false, true] { let arrays: Vec = vec![ Arc::new(Int32Array::from_iter_values(0..100)), @@ -3461,8 +3462,8 @@ mod tests { RecordBatch::try_new(schema.clone(), vec![array.clone(), array]).unwrap(); let bloom_props = |folding_enabled| { BloomFilterProperties::builder() - .with_max_ndv(1000) - .with_fpp(0.01) + .with_max_ndv(100_000) + .with_fpp(target_fpp) .with_folding_enabled(folding_enabled) .build() }; @@ -3490,7 +3491,7 @@ mod tests { let row_group = reader.get_row_group(0).unwrap(); let folded = row_group.get_column_bloom_filter(0).unwrap(); let retained = row_group.get_column_bloom_filter(1).unwrap(); - let mut expected = Sbbf::new_with_ndv_fpp(1000, 0.01).unwrap(); + let mut expected = Sbbf::new_with_ndv_fpp(100_000, target_fpp).unwrap(); for i in 0..100 { match batch.column(0).data_type() { DataType::Int32 => { @@ -3514,6 +3515,28 @@ mod tests { }; assert_eq!(serialize(retained), serialize(&expected)); assert!(serialize(folded).len() < serialize(retained).len()); + if batch.column(0).data_type() == &DataType::Int32 { + // Extra allocation can provide much stronger pruning than the requested + // FPP. Retaining it must preserve that benefit, not just its byte length. + let absent_values = 10_000_000_i32..11_000_000; + let folded_positives = + absent_values.clone().filter(|v| folded.check(v)).count(); + let retained_positives = absent_values.filter(|v| retained.check(v)).count(); + let mut folded_bits = Vec::new(); + let mut retained_bits = Vec::new(); + folded.write_bitset(&mut folded_bits).unwrap(); + retained.write_bitset(&mut retained_bits).unwrap(); + assert_eq!(folded_bits.len(), 128); + assert_eq!(retained_bits.len(), 128 * 1024); + assert_eq!(folded_positives, 9_088); + assert_eq!(retained_positives, 0); + let realized_fpp = folded_positives as f64 / 1_000_000.0; + assert!(realized_fpp <= target_fpp); + assert!( + folded_positives > retained_positives, + "folded={folded_positives}, retained={retained_positives}" + ); + } } } }