NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3999459704
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -665,8 +666,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_max_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);
+ let mut num_bits = calculated as i32;
+ let upper_bits = (BLOOM_FILTER_MAX_BYTES * 8) as i32;
+ if num_bits > upper_bits || calculated < 0.0 {
+ num_bits = upper_bits;
+ }
+ // This deliberately mirrors parquet-mr 1.17's integer expression,
including its unusual
+ // mask, so allocation thresholds remain compatible rather than merely
mathematically close.
+ num_bits = (num_bits + 255) & !256;
+ num_bits = num_bits.max((BLOOM_FILTER_MIN_BYTES * 8) as i32);
+ let requested = (num_bits as usize) / 8;
+ let allocated = requested
+ .clamp(BLOOM_FILTER_MIN_BYTES, BLOOM_FILTER_MAX_BYTES)
+ .next_power_of_two();
+ // The eligibility gate excludes max=32 when this cap would change
parquet-mr's allocation.
+ allocated.min(max_bytes)
+}
+
+/// Mirror the NDV/FPP sizing and power-of-two allocation used by the Apache
Arrow Rust `parquet`
+/// crate. Its source derives the formula from the standard Bloom-filter
false-positive equation
+/// with eight hash probes and links the underlying cache-efficient
Bloom-filter paper:
+///
https://github.com/apache/arrow-rs/blob/58.4.0/parquet/src/bloom_filter/mod.rs#L363-L395
+fn parquet_rs_bloom_filter_bytes(ndv: u64, fpp: f64) -> usize {
+ let num_bits =
+ (BLOOM_FILTER_HASH_PROBES * ndv as f64 /
bloom_filter_fpp_denominator(fpp)) as usize;
+ (num_bits / 8)
+ .clamp(BLOOM_FILTER_MIN_BYTES, BLOOM_FILTER_MAX_BYTES)
+ .next_power_of_two()
+}
+
+/// Encode an exact power-of-two allocation using parquet-rs 58.x's public
NDV/FPP setters.
+///
+/// A target `B > 32` is selected by every raw byte count in `(B/2, B]`. Aim
at `3B/4`, far from
+/// either floating-point boundary, and verify using the exact parquet-rs
sizing expression. The
+/// binary-search fallback covers unusual but still representable FPP values
without relying on
+/// the inverse formula landing on a particular floating-point integer.
+fn synthetic_ndv_for_bloom_filter_bytes(target_bytes: usize, fpp: f64) ->
DFResult<u64> {
+ let fpp_denominator = bloom_filter_fpp_denominator(fpp);
+ // Any raw size in (B / 2, B] rounds up to the target power-of-two
allocation B. Choose the
+ // midpoint of that interval to stay away from floating-point boundaries
at either end.
+ let raw_target_bytes = target_bytes as f64 * 3.0 / 4.0;
Review Comment:
Hi, thanks for review! No, this is not a copy-paste of any existing
codebase, but this is an implementation detail, and completely internal one -
does not affect user-facing behavior.
The rationale is that we don't have `max-bytes` property in parquet-rs, so
we're emulating the `max-bytes` capping logic of parquet-java using 'fake'
artificial NDV instead of missing `max-bytes` property. If user requested low
max-bytes limit, this limit would yield artificial NDV that is lower than
initially configured one, and parquet-rs would have to respect it. Thus, on the
outside this works exactly as max-bytes property would. This logic is
documented in the line you commented on:
> Choose the midpoint of that interval to stay away from floating-point
boundaries at either end.
Please tell me if its not clear enough.
IMO this should not be mentioned in any other docs. The only public contract
of this code is to guarantee that user-provided max-bytes is respected. The API
is well documented and limitation of max-bytes = power-of-two is documented too.
--
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]
