andygrove commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3973651665
##########
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;
+ let candidate = ((raw_target_bytes * fpp_denominator).round() as
u64).max(1);
+ if parquet_rs_bloom_filter_bytes(candidate, fpp) == target_bytes {
+ return Ok(candidate);
+ }
+
+ // Rust's standard binary-search helpers operate on materialized slices;
this is a lower-bound
+ // search over the implicit NDV domain `1..=u64::MAX`, so keep the numeric
search explicit.
+ let mut low = 1_u64;
+ let mut high = u64::MAX;
+ while low < high {
+ let mid = low + (high - low) / 2;
+ if parquet_rs_bloom_filter_bytes(mid, fpp) < target_bytes {
+ low = mid.saturating_add(1);
+ } else {
+ high = mid;
+ }
+ }
+ if parquet_rs_bloom_filter_bytes(low, fpp) == target_bytes {
+ Ok(low)
+ } else {
+ Err(DataFusionError::Internal(format!(
+ "FPP {fpp} cannot represent a {target_bytes}-byte parquet-rs Bloom
filter"
+ )))
+ }
+}
+
+fn validate_bloom_filter_inputs(fpp: f64, bytes: usize) -> DFResult<()> {
+ if !fpp.is_finite() || !(0.0..1.0).contains(&fpp) {
Review Comment:
Agreeing, and adding why this one has more teeth than it looks. In parquet
59.3.0 `set_column_bloom_filter_fpp` does not return an error for an
out-of-range value, it panics. `ColumnProperties::set_bloom_filter_fpp` calls
`validate_bloom_filter_fpp`, which rejects `!(fpp > 0.0 && fpp < 1.0)`, and the
caller does `panic!("{msg}")` (`properties.rs:1570` and `:1663`). So the
failure mode is an abort inside the native library across JNI, not a
differently sized filter.
It is latent today. I traced the only shape that reaches the setter with
`fpp == 0.0`: the denominator collapses to `-0.0`, so
`synthetic_ndv_for_bloom_filter_bytes` returns `Err` for every target except 32
bytes, which also needs `max-bytes=32`, and the JVM gate rejects an explicit
zero before either. Still, `fpp > 0.0 && fpp < 1.0` costs nothing and it is the
difference between a `DataFusionError` and a panic if that gate ever moves.
##########
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
Review Comment:
There is a third one outside the Rust file worth catching in the same pass.
The scaladoc on `requireNativeSupportedBloomFilterProperties` in
`CometIcebergNativeWrite.scala` also opens with "parquet-rs 58.x represents
Bloom filters as a power-of-two number of bytes", and
`synthetic_ndv_for_bloom_filter_bytes` says "using parquet-rs 58.x's public
NDV/FPP setters" while the code now calls the 59 setter.
##########
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);
Review Comment:
Measured this on Spark 4.1 with Iceberg 1.11 to put a number on it. A column
configured `enabled=true`, `fpp=0.01`, `ndv=1000`, with a single distinct value
inserted: the native writer emits a 32-byte filter and parquet-mr emits 2048, a
64x difference on an explicit-NDV column. Membership holds on both.
That matches what you and sunchao derived from source, and it confirms the
divergence is not confined to the no-NDV case the description frames it as.
--
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]
