This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new b7e5977133 [Parquet] Populate bloom filters from the dictionary while
a column is dictionary encoded (#10966)
b7e5977133 is described below
commit b7e5977133eadfe652eee4d4dc814a4c9d135533
Author: ranflarion <[email protected]>
AuthorDate: Wed Sep 9 10:52:06 2026 -0400
[Parquet] Populate bloom filters from the dictionary while a column is
dictionary encoded (#10966)
# Which issue does this PR close?
- Closes #10965.
- Related to #10963, which skips the filter for chunks whose data pages
are all dictionary encoded; the two are independent and merge in either
order.
# Rationale for this change
While a column is dictionary encoded every row was still hashed into the
bloom filter, although the interner already holds the distinct values.
For a low-cardinality column that is N inserts where D carry the same
information, and those are exactly the columns that stay dictionary
encoded. Suggested by @etseidl in #10963.
# What changes are included in this PR?
- `ColumnValueEncoderImpl::write_slice` and the byte array `encode`
insert into the filter only when no dictionary encoder is active.
- `flush_dict_page` on both encoders inserts every interned value before
handing the dictionary page over. It runs on fallback and at chunk
close, so after a fallback the filter holds the dictionary's values plus
every value written plain afterwards.
- `DictEncoder::uniques` exposes the interned values for the primitive
path.
The set of values inserted is unchanged and folding decides from the
final fill rate, so the serialized filter is byte-identical; only the
write-side cost changes.
Benchmark (`cargo bench -p parquet --bench arrow_writer --
'<batch>/bloom_filter'`, Apple M-series, criterion, main vs this
branch):
| batch | main | this PR | change |
|---|---|---|---|
| string_dictionary_low_cardinality_100 | 19.20 ms | 14.01 ms | -27.0% |
| primitive_non_null | 58.90 ms | 57.98 ms | -1.6% |
| string_dictionary | 48.85 ms | 49.10 ms | +0.5% (p = 0.24) |
| string_non_null | 108.63 ms | 108.96 ms | +0.3% (p = 0.05) |
# Are these changes tested?
Yes. New round-trip tests for `StringArray` and `Int64Array` cover a
chunk that stays dictionary encoded (asserted through the page encoding
mask) and a chunk that falls back to plain after a small dictionary page
limit, checking the filter for every written value and for absent ones.
The existing bloom filter round-trip tests already sweep dictionary
disabled, immediate fallback and dictionary enabled and pass unchanged.
# Are there any user-facing changes?
No API or output change. `DictEncoder::uniques` is new but the type is
not exported from the crate.
---
parquet/src/arrow/arrow_writer/byte_array.rs | 25 ++++--
parquet/src/arrow/arrow_writer/mod.rs | 105 ++++++++++++++++++++++++-
parquet/src/column/writer/encoder.rs | 24 ++++--
parquet/src/encodings/encoding/dict_encoder.rs | 5 ++
4 files changed, 142 insertions(+), 17 deletions(-)
diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs
b/parquet/src/arrow/arrow_writer/byte_array.rs
index e6f843f959..bd8ccbb36a 100644
--- a/parquet/src/arrow/arrow_writer/byte_array.rs
+++ b/parquet/src/arrow/arrow_writer/byte_array.rs
@@ -629,6 +629,13 @@ impl ColumnValueEncoder for ByteArrayEncoder {
));
}
+ if let Some(bloom_filter) = &mut self.bloom_filter {
+ let storage = encoder.interner.storage();
+ for range in &storage.values {
+ bloom_filter.insert(&storage.page[range.clone()]);
+ }
+ }
+
Ok(Some(encoder.flush_dict_page()))
}
_ => Ok(None),
@@ -679,16 +686,18 @@ where
}
}
- // encode the values into bloom filter if enabled
- if let Some(bloom_filter) = &mut encoder.bloom_filter {
- for idx in indices.clone() {
- bloom_filter.insert(values.value(idx).as_ref());
- }
- }
-
+ // While a dictionary is in use the filter is populated from its distinct
values in
+ // `flush_dict_page`, so each value is hashed once rather than once per
row.
match &mut encoder.dict_encoder {
Some(dict_encoder) => dict_encoder.encode(values, indices),
- None => encoder.fallback.encode(values, indices),
+ None => {
+ if let Some(bloom_filter) = &mut encoder.bloom_filter {
+ for idx in indices.clone() {
+ bloom_filter.insert(values.value(idx).as_ref());
+ }
+ }
+ encoder.fallback.encode(values, indices)
+ }
}
}
diff --git a/parquet/src/arrow/arrow_writer/mod.rs
b/parquet/src/arrow/arrow_writer/mod.rs
index 0bb19ffa11..2056922dd1 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -2107,7 +2107,7 @@ mod tests {
use num_traits::{FromPrimitive, ToPrimitive};
use tempfile::tempfile;
- use crate::basic::Encoding;
+ use crate::basic::{Encoding, EncodingMask};
use crate::data_type::AsBytes;
use crate::file::metadata::{ColumnChunkMetaData, ParquetMetaData,
ParquetMetaDataReader};
use crate::file::properties::{
@@ -3964,6 +3964,109 @@ mod tests {
);
}
+ fn write_with_bloom_filter(array: ArrayRef, dictionary_page_size_limit:
usize) -> Bytes {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "col",
+ array.data_type().clone(),
+ false,
+ )]));
+ let batch = RecordBatch::try_new(schema.clone(), vec![array]).unwrap();
+ let props = WriterProperties::builder()
+ .set_dictionary_enabled(true)
+ .set_dictionary_page_size_limit(dictionary_page_size_limit)
+ .set_write_batch_size(256)
+ .set_bloom_filter_enabled(true)
+ .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();
+ Bytes::from(buf)
+ }
+
+ fn data_page_encoding_mask(file: &Bytes) -> EncodingMask {
+ let metadata =
ParquetMetaDataReader::new().parse_and_finish(file).unwrap();
+ *metadata
+ .row_group(0)
+ .column(0)
+ .page_encoding_stats_mask()
+ .unwrap()
+ }
+
+ /// While a column is dictionary encoded the bloom filter is populated
from the dictionary
+ /// when it is flushed, so a chunk that stays dictionary encoded must
still contain every value.
+ #[test]
+ fn string_column_bloom_filter_populated_from_dictionary() {
+ let values: Vec<String> = (0..2000).map(|i| format!("value-{}", i %
10)).collect();
+ let array = Arc::new(StringArray::from_iter_values(&values));
+ let file = write_with_bloom_filter(array, 1024 * 1024);
+
assert!(data_page_encoding_mask(&file).is_only(Encoding::RLE_DICTIONARY));
+
+ check_bloom_filter(
+ vec![file],
+ "col".to_string(),
+ (0..10).map(|i| format!("value-{i}").into_bytes()).collect(),
+ (10..20)
+ .map(|i| format!("value-{i}").into_bytes())
+ .collect(),
+ );
+ }
+
+ /// After falling back from dictionary encoding the filter holds the
dictionary's values
+ /// and every value written plain afterwards.
+ #[test]
+ fn string_column_bloom_filter_across_dictionary_fallback() {
+ let values: Vec<String> = (0..2000).map(|i|
format!("value-{i}")).collect();
+ let array = Arc::new(StringArray::from_iter_values(&values));
+ let file = write_with_bloom_filter(array, 1024);
+ let encodings = data_page_encoding_mask(&file);
+ assert!(
+ encodings.is_set(Encoding::RLE_DICTIONARY) &&
encodings.is_set(Encoding::PLAIN),
+ "expected dictionary and plain data pages, got {encodings:?}"
+ );
+
+ check_bloom_filter(
+ vec![file],
+ "col".to_string(),
+ values.into_iter().map(String::into_bytes).collect(),
+ (2000..2010)
+ .map(|i| format!("value-{i}").into_bytes())
+ .collect(),
+ );
+ }
+
+ #[test]
+ fn i64_column_bloom_filter_populated_from_dictionary() {
+ let array = Arc::new(Int64Array::from_iter_values((0..2000).map(|i| i
% 10)));
+ let file = write_with_bloom_filter(array, 1024 * 1024);
+
assert!(data_page_encoding_mask(&file).is_only(Encoding::RLE_DICTIONARY));
+
+ check_bloom_filter(
+ vec![file],
+ "col".to_string(),
+ (0..10i64).collect(),
+ (10..20i64).collect(),
+ );
+ }
+
+ #[test]
+ fn i64_column_bloom_filter_across_dictionary_fallback() {
+ let array = Arc::new(Int64Array::from_iter_values(0..2000i64));
+ let file = write_with_bloom_filter(array, 1024);
+ let encodings = data_page_encoding_mask(&file);
+ assert!(
+ encodings.is_set(Encoding::RLE_DICTIONARY) &&
encodings.is_set(Encoding::PLAIN),
+ "expected dictionary and plain data pages, got {encodings:?}"
+ );
+
+ check_bloom_filter(
+ vec![file],
+ "col".to_string(),
+ (0..2000i64).collect(),
+ (2000..2010i64).collect(),
+ );
+ }
+
/// Test that bloom filter folding produces correct results even when
/// the configured NDV differs significantly from actual NDV.
/// A large NDV means a larger initial filter that gets folded down;
diff --git a/parquet/src/column/writer/encoder.rs
b/parquet/src/column/writer/encoder.rs
index 467c1510b5..0eadecb8b0 100644
--- a/parquet/src/column/writer/encoder.rs
+++ b/parquet/src/column/writer/encoder.rs
@@ -240,16 +240,18 @@ impl<T: DataType> ColumnValueEncoderImpl<T> {
}
}
- // encode the values into bloom filter if enabled
- if let Some(bloom_filter) = &mut self.bloom_filter {
- for value in slice {
- bloom_filter.insert(value);
- }
- }
-
+ // While a dictionary is in use the filter is populated from its
distinct values
+ // in `flush_dict_page`, so each value is hashed once rather than once
per row.
match &mut self.dict_encoder {
Some(encoder) => encoder.put(slice),
- _ => self.encoder.put(slice),
+ _ => {
+ if let Some(bloom_filter) = &mut self.bloom_filter {
+ for value in slice {
+ bloom_filter.insert(value);
+ }
+ }
+ self.encoder.put(slice)
+ }
}
}
}
@@ -413,6 +415,12 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
));
}
+ if let Some(bloom_filter) = &mut self.bloom_filter {
+ for value in encoder.uniques() {
+ bloom_filter.insert(value);
+ }
+ }
+
let buf = encoder.write_dict()?;
Ok(Some(DictionaryPage {
diff --git a/parquet/src/encodings/encoding/dict_encoder.rs
b/parquet/src/encodings/encoding/dict_encoder.rs
index 89666bbe73..995ade1f51 100644
--- a/parquet/src/encodings/encoding/dict_encoder.rs
+++ b/parquet/src/encodings/encoding/dict_encoder.rs
@@ -116,6 +116,11 @@ impl<T: DataType> DictEncoder<T> {
self.interner.storage().uniques.len()
}
+ /// Returns the distinct values interned so far, in dictionary order.
+ pub fn uniques(&self) -> &[T::T] {
+ &self.interner.storage().uniques
+ }
+
/// Returns size of unique values (keys) in the dictionary, in bytes.
pub fn dict_encoded_size(&self) -> usize {
self.interner.storage().size_in_bytes
