simplify

Author: adriangb

parquet/src/arrow/arrow_writer/byte_array.rs

@@ -17,7 +17,9 @@
 
 use crate::basic::Encoding;
 use crate::bloom_filter::Sbbf;
-use crate::column::writer::encoder::{ColumnValueEncoder, DataPageValues, DictionaryPage};
+use crate::column::writer::encoder::{
+    ColumnValueEncoder, DataPageValues, DictionaryPage, create_bloom_filter,
+};
 use crate::data_type::{AsBytes, ByteArray, Int32Type};
 use crate::encodings::encoding::{DeltaBitPackEncoder, Encoder};
 use crate::encodings::rle::RleEncoder;
@@ -423,7 +425,7 @@ pub struct ByteArrayEncoder {
     min_value: Option<ByteArray>,
     max_value: Option<ByteArray>,
     bloom_filter: Option<Sbbf>,
-    bloom_filter_target_fpp: Option<f64>,
+    bloom_filter_target_fpp: f64,
     geo_stats_accumulator: Option<Box<dyn GeoStatsAccumulator>>,
 }
 
@@ -432,9 +434,7 @@ impl ColumnValueEncoder for ByteArrayEncoder {
     type Values = dyn Array;
     fn flush_bloom_filter(&mut self) -> Option<Sbbf> {
         let mut sbbf = self.bloom_filter.take()?;
-        if let Some(target_fpp) = self.bloom_filter_target_fpp {
-            sbbf.fold_to_target_fpp(target_fpp);
-        }
+        sbbf.fold_to_target_fpp(self.bloom_filter_target_fpp);
         Some(sbbf)
     }
 
@@ -448,15 +448,7 @@ impl ColumnValueEncoder for ByteArrayEncoder {
 
         let fallback = FallbackEncoder::new(descr, props)?;
 
-        let (bloom_filter, bloom_filter_target_fpp) =
-            match props.bloom_filter_properties(descr.path()) {
-                Some(bf_props) => {
-                    let (sbbf, target_fpp) =
-                        Sbbf::from_properties(bf_props, props.max_row_group_row_count())?;
-                    (Some(sbbf), target_fpp)
-                }
-                None => (None, None),
-            };
+        let (bloom_filter, bloom_filter_target_fpp) = create_bloom_filter(props, descr)?;
 
         let statistics_enabled = props.statistics_enabled(descr.path());
 

parquet/src/bloom_filter/mod.rs

@@ -82,7 +82,6 @@ use crate::basic::{BloomFilterAlgorithm, BloomFilterCompression, BloomFilterHash
 use crate::data_type::AsBytes;
 use crate::errors::{ParquetError, Result};
 use crate::file::metadata::ColumnChunkMetaData;
-use crate::file::properties::{BloomFilterProperties, DEFAULT_MAX_ROW_GROUP_ROW_COUNT};
 use crate::file::reader::ChunkReader;
 use crate::parquet_thrift::{
     ElementType, FieldType, ReadThrift, ThriftCompactInputProtocol, ThriftCompactOutputProtocol,
@@ -205,14 +204,12 @@ impl std::ops::IndexMut<usize> for Block {
 /// parallel bit manipulation. When checking membership, only one block is accessed per query,
 /// eliminating the cache-miss penalty of standard Bloom filters.
 ///
-/// ## Two sizing modes
+/// ## Sizing and folding
 ///
-/// - **Fixed-size mode**: Created via [`Sbbf::new_with_ndv_fpp`] when the number of distinct
-///   values (NDV) is known. The filter is sized exactly for the given NDV and FPP.
-///
-/// - **Folding mode**: Created via [`Sbbf::new_with_num_of_bytes`] at a conservatively large
-///   size, then compacted after all values are inserted by calling [`Sbbf::fold_to_target_fpp`].
-///   This eliminates the need to know NDV upfront.
+/// Filters are initially sized for a maximum expected number of distinct values (NDV) via
+/// [`Sbbf::new_with_ndv_fpp`]. After all values are inserted, the filter is compacted by
+/// calling [`Sbbf::fold_to_target_fpp`], which folds the filter down to the smallest size
+/// that still meets the target false positive probability.
 ///
 /// The creation of this structure is based on the [`crate::file::properties::BloomFilterProperties`]
 /// struct set via [`crate::file::properties::WriterProperties`] and is thus hidden by default.
@@ -276,34 +273,6 @@ pub(crate) fn num_of_bits_from_ndv_fpp(ndv: u64, fpp: f64) -> usize {
 }
 
 impl Sbbf {
-    /// Create a bloom filter from [`BloomFilterProperties`], returning the filter and an
-    /// optional target FPP for folding mode.
-    ///
-    /// | `ndv` | `max_bytes` | Behavior |
-    /// |-------|-------------|----------|
-    /// | Set   | _           | Fixed-size filter sized for that NDV (legacy). Returns `None` for FPP. |
-    /// | None  | Set         | Filter of `max_bytes`, folded at flush via [`Sbbf::fold_to_target_fpp`]. |
-    /// | None  | None        | Filter sized for `max_distinct_items` (default: [`DEFAULT_MAX_ROW_GROUP_ROW_COUNT`]) items at the given FPP, folded at flush. |
-    pub fn from_properties(
-        props: &BloomFilterProperties,
-        max_distinct_items: Option<usize>,
-    ) -> Result<(Self, Option<f64>)> {
-        match props.ndv {
-            Some(ndv) => {
-                // Fixed-size mode: size based on explicit NDV (legacy behavior)
-                Ok((Self::new_with_ndv_fpp(ndv, props.fpp)?, None))
-            }
-            None => {
-                // Folding mode: allocate large, fold down at flush
-                let max_bytes = props.max_bytes.unwrap_or_else(|| {
-                    let ndv = max_distinct_items.unwrap_or(DEFAULT_MAX_ROW_GROUP_ROW_COUNT) as u64;
-                    num_of_bits_from_ndv_fpp(ndv, props.fpp) / 8
-                });
-                Ok((Self::new_with_num_of_bytes(max_bytes), Some(props.fpp)))
-            }
-        }
-    }
-
     /// Create a new [Sbbf] with given number of distinct values and false positive probability.
     /// Will return an error if `fpp` is greater than or equal to 1.0 or less than 0.0.
     pub fn new_with_ndv_fpp(ndv: u64, fpp: f64) -> Result<Self, ParquetError> {

parquet/src/column/writer/encoder.rs

@@ -27,7 +27,9 @@ use crate::data_type::DataType;
 use crate::data_type::private::ParquetValueType;
 use crate::encodings::encoding::{DictEncoder, Encoder, get_encoder};
 use crate::errors::{ParquetError, Result};
-use crate::file::properties::{EnabledStatistics, WriterProperties};
+use crate::file::properties::{
+    DEFAULT_MAX_ROW_GROUP_ROW_COUNT, EnabledStatistics, WriterProperties,
+};
 use crate::geospatial::accumulator::{GeoStatsAccumulator, try_new_geo_stats_accumulator};
 use crate::geospatial::statistics::GeospatialStatistics;
 use crate::schema::types::{ColumnDescPtr, ColumnDescriptor};
@@ -138,7 +140,7 @@ pub struct ColumnValueEncoderImpl<T: DataType> {
     min_value: Option<T::T>,
     max_value: Option<T::T>,
     bloom_filter: Option<Sbbf>,
-    bloom_filter_target_fpp: Option<f64>,
+    bloom_filter_target_fpp: f64,
     variable_length_bytes: Option<i64>,
     geo_stats_accumulator: Option<Box<dyn GeoStatsAccumulator>>,
 }
@@ -189,9 +191,7 @@ impl<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> {
 
     fn flush_bloom_filter(&mut self) -> Option<Sbbf> {
         let mut sbbf = self.bloom_filter.take()?;
-        if let Some(target_fpp) = self.bloom_filter_target_fpp {
-            sbbf.fold_to_target_fpp(target_fpp);
-        }
+        sbbf.fold_to_target_fpp(self.bloom_filter_target_fpp);
         Some(sbbf)
     }
 
@@ -210,15 +210,7 @@ impl<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> {
 
         let statistics_enabled = props.statistics_enabled(descr.path());
 
-        let (bloom_filter, bloom_filter_target_fpp) =
-            match props.bloom_filter_properties(descr.path()) {
-                Some(bf_props) => {
-                    let (sbbf, target_fpp) =
-                        Sbbf::from_properties(bf_props, props.max_row_group_row_count())?;
-                    (Some(sbbf), target_fpp)
-                }
-                None => (None, None),
-            };
+        let (bloom_filter, bloom_filter_target_fpp) = create_bloom_filter(props, descr)?;
 
         let geo_stats_accumulator = try_new_geo_stats_accumulator(descr);
 
@@ -395,6 +387,28 @@ fn replace_zero<T: ParquetValueType>(val: &T, descr: &ColumnDescriptor, replace:
     }
 }
 
+/// Creates a bloom filter sized for the column's configured NDV (or the row group size
+/// if NDV is not explicitly set), returning the filter and the target FPP for folding.
+pub(crate) fn create_bloom_filter(
+    props: &WriterProperties,
+    descr: &ColumnDescPtr,
+) -> Result<(Option<Sbbf>, f64)> {
+    match props.bloom_filter_properties(descr.path()) {
+        Some(bf_props) => {
+            let ndv = bf_props.ndv.unwrap_or(
+                props
+                    .max_row_group_row_count()
+                    .unwrap_or(DEFAULT_MAX_ROW_GROUP_ROW_COUNT) as u64,
+            );
+            Ok((
+                Some(Sbbf::new_with_ndv_fpp(ndv, bf_props.fpp)?),
+                bf_props.fpp,
+            ))
+        }
+        None => Ok((None, 0.0)),
+    }
+}
+
 fn update_geo_stats_accumulator<'a, T, I>(bounder: &mut dyn GeoStatsAccumulator, iter: I)
 where
     T: ParquetValueType + 'a,

parquet/src/file/properties.rs

@@ -54,8 +54,7 @@ pub const DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH: Option<usize> = Some(64);
 /// Default value for [`BloomFilterProperties::fpp`]
 pub const DEFAULT_BLOOM_FILTER_FPP: f64 = 0.05;
 /// Default value for [`BloomFilterProperties::ndv`]
-#[deprecated(note = "NDV is now optional; bloom filters use folding mode by default")]
-pub const DEFAULT_BLOOM_FILTER_NDV: u64 = 1_000_000_u64;
+pub const DEFAULT_BLOOM_FILTER_NDV: u64 = DEFAULT_MAX_ROW_GROUP_ROW_COUNT as u64;
 /// Default values for [`WriterProperties::statistics_truncate_length`]
 pub const DEFAULT_STATISTICS_TRUNCATE_LENGTH: Option<usize> = Some(64);
 /// Default value for [`WriterProperties::offset_index_disabled`]
@@ -997,11 +996,13 @@ impl WriterPropertiesBuilder {
         self
     }
 
-    /// Sets default number of distinct values (ndv) for bloom filter for all columns.
+    /// Sets default maximum expected number of distinct values (ndv) for bloom filter
+    /// for all columns (defaults to [`DEFAULT_BLOOM_FILTER_NDV`]).
     ///
-    /// When set, this activates fixed-size mode: the bloom filter is sized exactly for
-    /// the given NDV at the configured FPP, with no folding. When not set (default),
-    /// the bloom filter uses folding mode instead.
+    /// The bloom filter is initially sized for this many distinct values at the
+    /// configured FPP, then folded down after all values are inserted to achieve
+    /// optimal size. A good heuristic is to set this to the expected number of rows
+    /// in the row group.
     ///
     /// Implicitly enables bloom writing, as if [`set_bloom_filter_enabled`] had
     /// been called.
@@ -1012,26 +1013,6 @@ impl WriterPropertiesBuilder {
         self
     }
 
-    /// Sets the default maximum initial allocation size in bytes for bloom filter folding mode
-    /// for all columns.
-    ///
-    /// When bloom filters use folding mode (no explicit NDV), this controls the initial
-    /// allocation size. The filter will be folded down at flush time to meet the target FPP.
-    /// If not set, the initial size is derived from `max_row_group_row_count` and `fpp`.
-    ///
-    /// The value will be rounded up to the next power of two, bounded by
-    /// [`BITSET_MIN_LENGTH`](crate::bloom_filter::BITSET_MIN_LENGTH) and
-    /// [`BITSET_MAX_LENGTH`](crate::bloom_filter::BITSET_MAX_LENGTH).
-    ///
-    /// Implicitly enables bloom writing, as if [`set_bloom_filter_enabled`] had been called.
-    ///
-    /// [`set_bloom_filter_enabled`]: Self::set_bloom_filter_enabled
-    pub fn set_bloom_filter_max_bytes(mut self, value: usize) -> Self {
-        self.default_column_properties
-            .set_bloom_filter_max_bytes(value);
-        self
-    }
-
     // ----------------------------------------------------------------------
     // Setters for a specific column
 
@@ -1137,15 +1118,6 @@ impl WriterPropertiesBuilder {
         self.get_mut_props(col).set_bloom_filter_ndv(value);
         self
     }
-
-    /// Sets the maximum initial allocation size in bytes for bloom filter folding mode
-    /// for a specific column.
-    ///
-    /// Takes precedence over [`Self::set_bloom_filter_max_bytes`].
-    pub fn set_column_bloom_filter_max_bytes(mut self, col: ColumnPath, value: usize) -> Self {
-        self.get_mut_props(col).set_bloom_filter_max_bytes(value);
-        self
-    }
 }
 
 impl From<WriterProperties> for WriterPropertiesBuilder {
@@ -1225,15 +1197,12 @@ impl Default for EnabledStatistics {
 
 /// Controls the bloom filter to be computed by the writer.
 ///
-/// Two modes are supported:
+/// 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.
 ///
-/// - **Fixed-size mode**: When `ndv` is set to `Some(n)`, the bloom filter is sized based on `ndv`
-///   and `fpp` at allocation time. This is the legacy behavior.
-///
-/// - **Folding mode** (default): When `ndv` is `None`, a conservatively large bloom filter is
-///   allocated (sized for worst-case NDV = max row group rows, or `max_bytes` if set), then
-///   folded down at flush time to meet the target `fpp`. This eliminates the need to guess NDV
-///   upfront and produces optimally-sized filters automatically.
+/// [`Sbbf::fold_to_target_fpp`]: crate::bloom_filter::Sbbf::fold_to_target_fpp
 #[derive(Debug, Clone, PartialEq)]
 pub struct BloomFilterProperties {
     /// False positive probability. This should be always between 0 and 1 exclusive. Defaults to [`DEFAULT_BLOOM_FILTER_FPP`].
@@ -1242,32 +1211,28 @@ pub struct BloomFilterProperties {
     ///
     /// The bloom filter data structure is a trade of between disk and memory space versus fpp, the
     /// smaller the fpp, the more memory and disk space is required, thus setting it to a reasonable value
-    /// e.g. 0.1, 0.01, or 0.001 is recommended.
+    /// 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.
     pub fpp: f64,
-    /// Number of distinct values. When set to `Some(n)`, the bloom filter is sized exactly for
-    /// `n` distinct values at the given `fpp` (fixed-size mode). When `None` (default), the
-    /// filter uses folding mode instead.
+    /// Maximum expected number of distinct values. When `None` (default), the bloom filter
+    /// is sized based on the row group's `max_row_group_row_count` at runtime.
     ///
     /// You should set this value by calling [`WriterPropertiesBuilder::set_bloom_filter_ndv`].
     ///
-    /// Usage of bloom filter is most beneficial for columns with large cardinality, so a good heuristic
-    /// is to set ndv to the number of rows. However, it can reduce disk size if you know in advance a smaller
-    /// number of distinct values.
+    /// 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 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.
     pub ndv: Option<u64>,
-    /// Maximum initial allocation size in bytes for folding mode. When `None` (default), the
-    /// initial size is derived from `max_row_group_row_count` and `fpp`. Only used when `ndv`
-    /// is `None`.
-    ///
-    /// You should set this value by calling [`WriterPropertiesBuilder::set_bloom_filter_max_bytes`].
-    pub max_bytes: Option<usize>,
 }
 
 impl Default for BloomFilterProperties {
     fn default() -> Self {
         BloomFilterProperties {
             fpp: DEFAULT_BLOOM_FILTER_FPP,
             ndv: None,
-            max_bytes: None,
         }
     }
 }
@@ -1364,22 +1329,14 @@ impl ColumnProperties {
             .fpp = value;
     }
 
-    /// Sets the number of distinct (unique) values for bloom filter for this column, and implicitly
-    /// enables bloom filter if not previously enabled. This activates fixed-size mode (no folding).
+    /// Sets the maximum expected number of distinct (unique) values for bloom filter for this
+    /// column, and implicitly enables bloom filter if not previously enabled.
     fn set_bloom_filter_ndv(&mut self, value: u64) {
         self.bloom_filter_properties
             .get_or_insert_with(Default::default)
             .ndv = Some(value);
     }
 
-    /// Sets the maximum initial allocation size in bytes for bloom filter folding mode, and
-    /// implicitly enables bloom filter if not previously enabled.
-    fn set_bloom_filter_max_bytes(&mut self, value: usize) {
-        self.bloom_filter_properties
-            .get_or_insert_with(Default::default)
-            .max_bytes = Some(value);
-    }
-
     /// Returns optional encoding for this column.
     fn encoding(&self) -> Option<Encoding> {
         self.encoding
@@ -1723,7 +1680,6 @@ mod tests {
                 Some(&BloomFilterProperties {
                     fpp: 0.1,
                     ndv: Some(100),
-                    max_bytes: None,
                 })
             );
         }
@@ -1762,7 +1718,6 @@ mod tests {
             Some(&BloomFilterProperties {
                 fpp: DEFAULT_BLOOM_FILTER_FPP,
                 ndv: None,
-                max_bytes: None,
             })
         );
     }
@@ -1806,7 +1761,6 @@ mod tests {
             Some(&BloomFilterProperties {
                 fpp: DEFAULT_BLOOM_FILTER_FPP,
                 ndv: Some(100),
-                max_bytes: None,
             })
         );
         assert_eq!(
@@ -1817,7 +1771,6 @@ mod tests {
             Some(&BloomFilterProperties {
                 fpp: 0.1,
                 ndv: None,
-                max_bytes: None,
             })
         );
     }