unikdahal commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3951978712
##########
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);
+ 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();
+ // Unlike parquet-mr's strict-bound bug at exactly 32 bytes, honor
Iceberg's configured cap.
+ allocated.min(max_bytes)
Review Comment:
I don't think we should intentionally change the `max-bytes=32` behavior
while describing this sizing logic as parquet-mr compatible.
In parquet-mr, `BlockSplitBloomFilter` only installs `maximumBytes` when it
is strictly greater than the 32-byte lower bound. So a configured maximum of
exactly 32 is effectively not used as the maximum when NDV/FPP request a larger
filter.
For example:
`NDV=1,000,000`, `FPP=0.0001`, `max-bytes=32`
requests about 2.63 MiB before power-of-two rounding, and the JVM writer
ends up with a 4 MiB Bloom filter because the 32-byte maximum is ignored. This
implementation forcibly returns 32 bytes instead — a very large pruning-quality
difference.
Could we either emulate the parquet-mr behavior here or conservatively fall
back to the classic writer for `max-bytes=32` when it matters? The existing
binding-cap test would be stronger if it compared the JVM and native footer
sizes for this boundary.
##########
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:
There is one more parquet-mr parity edge case here for very large but still
valid NDVs.
parquet-mr calculates `-8 * n` using Java `long` arithmetic before the
division/conversion to floating point. That multiplication can overflow. Here
`ndv` is converted to `f64` before multiplication, so the Java overflow
behavior can never occur.
For example, with `NDV = 2^61`, Java's `-8 * n` wraps to zero and parquet-mr
ends up requesting the minimum 32-byte Bloom filter, while this implementation
calculates a huge value and caps it at the configured maximum (1 MiB by
default).
Since planning currently accepts the full positive Java `long` range, this
can silently select native execution with materially different output. Rather
than reproducing the overflow, could we conservatively fall back for `ndv >
Long.MaxValue / 8` and add tests at the threshold, threshold + 1, `2^61`, and
`Long.MaxValue`?
##########
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala:
##########
@@ -100,10 +116,36 @@ object IcebergWriteProtoTranslation {
}
}
+ private def enabledBloomFilterColumnNames(props: Map[String, String]):
Seq[String] =
+ props.iterator
+ .collect {
+ case (key, value)
+ if key.startsWith(Keys.ParquetBloomFilterColumnEnabledPrefix) &&
+ value.equalsIgnoreCase("true") =>
Review Comment:
Could we avoid limiting this to only `enabled=true` columns?
Iceberg iterates every entry under the enabled-column prefix and applies the
settings in this order: `withBloomFilterEnabled`, then FPP, then NDV.
parquet-mr's `withBloomFilterNDV` explicitly sets `bloomFilterEnabled=true`.
That means:
`enabled=false` + `ndv=<positive value>`
still produces a Bloom filter on the JVM path. Here the column is dropped
completely, so native writes no Bloom filter. The same filtering also means
associated FPP/NDV values are not parsed/validated when `enabled=false`, which
can make native succeed where the JVM path would reject a malformed value.
Could we model the effective state using Iceberg's actual application
ordering rather than treating literal `true` as the complete set of
Bloom-filter columns, and add JVM/native parity tests for `false + NDV` and
malformed FPP/NDV in that combination?
--
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]
