feat(parquet): allow disabling Bloom filter folding

parquet/src/arrow/arrow_writer/byte_array.rs

@@ -429,7 +429,7 @@ pub struct ByteArrayEncoder {
     min_value: Option<ByteArray>,
     max_value: Option<ByteArray>,
     bloom_filter: Option<Sbbf>,
-    bloom_filter_target_fpp: f64,
+    bloom_filter_target_fpp: Option<f64>,
     geo_stats_accumulator: Option<Box<dyn GeoStatsAccumulator>>,
     is_decimal: bool,
 }
@@ -439,7 +439,9 @@ impl ColumnValueEncoder for ByteArrayEncoder {
     type Values = dyn Array;
     fn flush_bloom_filter(&mut self) -> Option<Sbbf> {
         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)
     }
 

parquet/src/arrow/arrow_writer/mod.rs

@@ -3440,6 +3440,107 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bloom_filter_folding_can_be_disabled_per_column() {
+        use crate::bloom_filter::Sbbf;
+        use crate::file::properties::BloomFilterProperties;
+
+        let target_fpp = 0.01;
+        for dictionary_enabled in [false, true] {
+            let arrays: Vec<ArrayRef> = 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(100_000)
+                        .with_fpp(target_fpp)
+                        .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(100_000, target_fpp).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());
+                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}"
+                    );
+                }
+            }
+        }
+    }
+
     fn write_with_bloom_filter(array: ArrayRef, dictionary_page_size_limit: usize) -> Bytes {
         let schema = Arc::new(Schema::new(vec![Field::new(
             "col",

parquet/src/column/writer/encoder.rs

@@ -205,7 +205,7 @@ pub struct ColumnValueEncoderImpl<T: DataType> {
     max_value: Option<T::T>,
     nan_count: Option<u64>,
     bloom_filter: Option<Sbbf>,
-    bloom_filter_target_fpp: f64,
+    bloom_filter_target_fpp: Option<f64>,
     variable_length_bytes: Option<i64>,
     geo_stats_accumulator: Option<Box<dyn GeoStatsAccumulator>>,
 }
@@ -261,7 +261,9 @@ impl<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> {
 
     fn flush_bloom_filter(&mut self) -> Option<Sbbf> {
         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<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> {
 }
 
 /// 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<Sbbf>, f64)> {
+) -> Result<(Option<Sbbf>, Option<f64>)> {
     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)),
     }
 }
 

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,13 +1579,15 @@ impl Default for EnabledStatistics {
 pub struct BloomFilterProperties {
     fpp: f64,
     ndv: u64,
+    folding_enabled: bool,
 }
 
 impl Default for BloomFilterProperties {
     fn default() -> Self {
         BloomFilterProperties {
             fpp: DEFAULT_BLOOM_FILTER_FPP,
             ndv: DEFAULT_BLOOM_FILTER_NDV,
+            folding_enabled: true,
         }
     }
 }
@@ -1605,24 +1608,32 @@ 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`].
     ///
     /// 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<f64>,
     ndv: Option<u64>,
+    folding_enabled: Option<bool>,
 }
 
 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,13 +2653,15 @@ mod tests {
             Some(&BloomFilterProperties {
                 fpp: DEFAULT_BLOOM_FILTER_FPP,
                 ndv: 100,
+                ..Default::default()
             })
         );
         assert_eq!(
             props.bloom_filter_properties(&col),
             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();