NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3952603315
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -655,8 +656,146 @@ fn build_writer_properties(settings:
&IcebergParquetWriteSettings) -> DFResult<W
.set_dictionary_page_size_limit(settings.dict_size_bytes as usize)
.set_data_page_row_count_limit(settings.page_row_limit as usize)
.set_statistics_enabled(EnabledStatistics::Page)
- .set_statistics_truncate_length(None)
- .build())
+ .set_statistics_truncate_length(None);
+ for column in &settings.bloom_filter_enabled_columns {
+ let path = parquet_column_path(column);
+ let fpp = settings
+ .bloom_filter_fpp_by_column
+ .get(column)
+ .copied()
+ .unwrap_or(ICEBERG_DEFAULT_BLOOM_FILTER_FPP);
+ let ndv = settings.bloom_filter_ndv_by_column.get(column).copied();
+ let max_bytes = settings.bloom_filter_max_bytes as usize;
+ validate_bloom_filter_inputs(fpp, max_bytes)?;
+ let target_bytes = parquet_mr_bloom_filter_bytes(ndv, fpp, max_bytes);
+ let synthetic_ndv = synthetic_ndv_for_bloom_filter_bytes(target_bytes,
fpp)?;
+ builder = builder
+ .set_column_bloom_filter_enabled(path.clone(), true)
+ .set_column_bloom_filter_fpp(path.clone(), fpp)
+ .set_column_bloom_filter_ndv(path, synthetic_ndv);
+ }
+ Ok(builder.build())
+}
+
+/// Convert the dot-separated physical path supplied by Iceberg Java into
parquet-rs path parts.
+/// `ColumnPath::from(&str)` creates one literal part and therefore cannot
represent nested leaves.
+fn parquet_column_path(path: &str) -> ColumnPath {
+ ColumnPath::from(path.split('.').map(str::to_owned).collect::<Vec<_>>())
+}
+
+// Match Apache Parquet Java's BlockSplitBloomFilter implementation bounds:
+//
https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L40-L50
+const BLOOM_FILTER_MIN_BYTES: usize = 32;
+const BLOOM_FILTER_MAX_BYTES: usize = 128 * 1024 * 1024;
+const BLOOM_FILTER_HASH_PROBES: f64 = 8.0;
+const ICEBERG_DEFAULT_BLOOM_FILTER_FPP: f64 = 0.01;
+#[cfg(test)]
+const ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES: usize = 1024 * 1024;
+
+/// The positive denominator obtained by solving the Bloom-filter
false-positive equation
+/// `fpp = (1 - exp(-k * ndv / bits))^k` for `bits`, with the Parquet SBBF's
`k = 8` probes.
+///
+/// See the Apache Arrow Rust `parquet` implementation and its cited paper:
+///
https://github.com/apache/arrow-rs/blob/58.4.0/parquet/src/bloom_filter/mod.rs#L369-L376
+/// http://algo2.iti.kit.edu/documents/cacheefficientbloomfilters-jea.pdf
+fn bloom_filter_fpp_denominator(fpp: f64) -> f64 {
+ -(1.0 - fpp.powf(1.0 / BLOOM_FILTER_HASH_PROBES)).ln()
+}
+
+/// Reproduce parquet-mr's non-adaptive allocation decision before translating
the resulting
+/// power-of-two byte size into parquet-rs's NDV-shaped API. An absent NDV
requests the full cap;
+/// an explicit NDV sizes from NDV/FPP and then applies the cap. The native
eligibility gate only
+/// admits representable power-of-two caps.
+///
+/// Apache Parquet Java implementation:
+///
https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L277-L301
+///
https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L195-L218
+fn parquet_mr_bloom_filter_bytes(ndv: Option<u64>, fpp: f64, max_bytes: usize)
-> usize {
+ let Some(ndv) = ndv else {
+ return max_bytes;
+ };
+
+ let calculated = BLOOM_FILTER_HASH_PROBES * ndv as f64 /
bloom_filter_fpp_denominator(fpp);
Review Comment:
This does not look like a feature one can actually rely upon. But I guess it
doesn't hurt to have this behavior replicated here. Merged in
[7d218de](https://github.com/apache/datafusion-comet/pull/5724/commits/7d218de7c92536f27ae872c5fdcf52fcd4b1c4e8).
IMO so many distinct values in a column is not a reasonable usecase for
bloom filters anyway, so producing a tiny 32b "placeholder" filter for such
columns is even better implementation choice than actually allocating 128mb.
That said, I would even lower this "NDV max limit" down from 2^60 to something
more realistic if I could.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
