diff --git a/docs/source/contributor-guide/benchmarking.md b/docs/source/contributor-guide/benchmarking.md
index c3318515e4f..420f78abd9f 100644
--- a/docs/source/contributor-guide/benchmarking.md
+++ b/docs/source/contributor-guide/benchmarking.md
@@ -43,6 +43,47 @@ These can also be run as a suite on a dedicated machine, see
[Micro Benchmarking on AWS EC2](benchmarking_micro_ec2.md). Published results are in
[benchmarks/results/micro](https://github.com/apache/datafusion-comet/tree/main/benchmarks/results/micro).
+## Iceberg Bloom-filter writes
+
+`CometIcebergWriteBenchmark` compares Spark and Comet Iceberg writers. Its `bloom` mode compares
+native writes with Bloom filters off/on (requiring Iceberg 1.10+), using the same prepared Parquet source in eight files. Both cases enable native writes and the Bloom folding opt-in; only the table's Bloom
+properties differ. Filters cover the bigint, string, and binary columns at the default 1 MiB cap,
+without explicit NDV/FPP. The benchmark checks the native plan, row count, and filter presence
+before measuring. Table creation and source preparation are outside the timed insert; the Iceberg
+commit is included.
+
+Use the existing benchmark target for timing, optionally followed by rows and minimum iterations:
+
+```shell
+make benchmark-org.apache.spark.sql.benchmark.CometIcebergWriteBenchmark bloom
+```
+
+Each case warms up and measures at least five writes of 4,194,304 rows by default. Use
+`bloom-off` or `bloom-on` to select a single case, or append `32768 1` for a quick functional
+check. Performance comparisons require a release build.
+
+To collect peak RSS, build once and run each case in a separate JVM using the existing Maven
+invocation. Obtain the shared JVM options from the Makefile:
+
+```shell
+make release PROFILES=-Pspark-4.1
+export COMET_CONF_DIR="$PWD/conf"
+export MAVEN_OPTS="$(make --no-print-directory print-benchmark-args PROFILES=-Pspark-4.1 BENCH_HEAP=8g | sed -n 's/^MAVEN_OPTS=//p')"
+cd spark
+/usr/bin/time -v ../mvnw exec:java -Dexec.classpathScope=test \
+ -Dexec.cleanupDaemonThreads=false \
+ -Dexec.mainClass=org.apache.spark.sql.benchmark.CometIcebergWriteBenchmark \
+ -Dexec.args="bloom-off" -Pspark-4.1
+```
+
+Repeat with `bloom-on`. On macOS, use `/usr/bin/time -l` instead of `-v`; its maximum RSS is in
+bytes, while GNU time reports KiB. The reported RSS includes Maven, source generation, validation,
+warmup, and writes; it is not the writer's allocation or its peak alone. Fresh processes avoid
+carrying an earlier case's RSS high-water mark and retained native buffers into the next case.
+The warehouse is local, so these results do not include object-store latency.
+
+Do not measure RSS around `make benchmark-*`: its release-build prerequisite includes compilation.
+
```{toctree}
:hidden:
diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md
index 373f3c91df5..a299f2b2cba 100644
--- a/docs/source/contributor-guide/iceberg-writes.md
+++ b/docs/source/contributor-guide/iceberg-writes.md
@@ -458,6 +458,11 @@ The Comet suites run against the Iceberg version each Spark profile pins in `spa
for Spark 3.4, 1.8.1 for 3.5, 1.10.0 for 4.0, and 1.11.0 for 4.1. No Iceberg runtime is published
for Spark 4.2, so the 4.2 profile builds against the 4.0 runtime and the Iceberg suites skip there
(`icebergAvailable` is false on Spark 4.2). Only the default profile runs on every pull request.
+
+Bloom-enabled native writes additionally require an identifiable Iceberg version >= 1.10.
+On the 3.4 and 3.5 profiles, any per-column Bloom property (including `false` enablement and standalone FPP/NDV) selects Java,
+even when the Bloom folding opt-in is enabled. Other native-write eligibility is unchanged.
+
Behaviour that differs between Iceberg versions, such as reflection targets, metrics conventions
and partition path spelling, needs a test that runs on each.
diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index 972ceea6fbf..bca71b660db 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -485,9 +485,11 @@ A write is eligible only when ALL of the following hold:
| `write.parquet.page-version` | unset or `v1` |
| `write.parquet.shred-variants` | unset or `false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write) |
| `write.parquet.variant-inference-buffer-size` | any value (only meaningful when shredding, which is gated) |
-| `write.parquet.bloom-filter-enabled.column.` | unset or `false` |
| `write.metadata.metrics.*` | any value (manifest metrics are re-derived on the JVM with Iceberg's own logic) |
| `write.spark.fanout.enabled` | any value (the native writer implements both clustered and fanout modes) |
+| `write.parquet.bloom-filter-enabled.column.` | `true` or `false`; native eligibility requires identifiable Iceberg 1.10+. Older or unidentified runtimes use Java for any per-column Bloom property, including `false` enablement and standalone FPP/NDV. On Iceberg 1.11+, an explicit NDV enables the column even when this value is `false`. Enabled columns use the native writer when `spark.comet.write.iceberg.bloomFilterFolding.enabled=true` |
+| `write.parquet.bloom-filter-fpp.column.` / `write.parquet.bloom-filter-ndv.column.` | Iceberg 1.10 interprets FPP but ignores NDV; 1.11 interprets both. For interpreted properties on a column named by an `enabled` property, FPP is finite and strictly between 0 and 1 (default `0.01`), and NDV is a positive Java long no greater than `Long.MAX_VALUE / 8`. |
+| `write.parquet.bloom-filter-max-bytes` | `write.parquet.bloom-filter-enabled.column.*` present (even `false`): unset (1 MiB) or a power of two from 32 bytes through 128 MiB; larger explicit NDV/FPP requests fall back at 32 bytes because Java ignores that cap. With no such key: unset or any positive Java int, including `100`. Malformed, zero, or negative caps use Java and fail its validation. |
| `write.target-file-size-bytes` | any value (the two writers can choose different roll points; see accepted divergences) |
| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs`, matched case-sensitively (`S3://` falls back). `s3`, `s3a` and `gs` need a bucket in the authority (`s3://bucket/...`), so a hostless form such as `s3:/bucket/key` falls back. `gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below |
| resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` |
@@ -498,7 +500,7 @@ A write is eligible only when ALL of the following hold:
Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` —
everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g.
-`bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions),
+`stats-enabled.column.*`, keys added by future Iceberg versions),
any `parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*`
key in the session Hadoop configuration other than the reader-only
`parquet.hadoop.vectored.io.enabled` (with `HadoopFileIO`-backed output those reach
@@ -653,13 +655,77 @@ unchanged.
### Accepted divergences
-Some differences between parquet-mr and the pinned parquet-rs / iceberg-rust are unconditional —
-they apply to every native write and cannot be configured away. Enabling
-`spark.comet.write.iceberg.enabled` accepts them. They fall into two classes with very
-different blast radius: differences confined to the physical bytes of a data file (cosmetic —
-no reader decision is based on them), and differences visible in manifest metadata (these
-outlive the write and feed later readers' pruning decisions, so each one is analyzed
-individually below).
+Enabling `spark.comet.write.iceberg.enabled` accepts the physical file-layout and manifest-metadata
+differences described below. Bloom-filter differences require the additional
+`spark.comet.write.iceberg.bloomFilterFolding.enabled` opt-in, because they can affect pruning.
+
+#### Bloom filters
+
+Comet's native Iceberg writer supports tables with column Bloom filters. However, byte-identical
+behavior to classic Spark cannot be guaranteed for all settings: parquet-rs enables folding by
+default, while Spark's non-adaptive parquet-java writer does not.
+
+Folding automatically shrinks a filter when its initial allocation from NDV or `max-bytes` is
+larger than the actual values need at the requested false-positive probability (FPP). It saves
+serialized space by shrinking toward the target FPP; Java retains the larger filter. Folding
+retains the original buffer capacity, so this does not imply lower peak writer memory.
+
+In addition to the [general native-write eligibility conditions](#native-parquet-write-eligibility),
+native writes for configured Bloom columns require all of the following:
+
+- `spark.comet.write.iceberg.enabled=true`.
+- `spark.comet.write.iceberg.bloomFilterFolding.enabled=true`.
+- The Iceberg runtime reports an identifiable version of 1.10 or later. Earlier or
+ unidentified runtimes use Java whenever a per-column Bloom property is configured,
+ including `false` enablement and standalone FPP/NDV properties. The opt-in cannot enable native Bloom writes there.
+- The effective `write.parquet.bloom-filter-max-bytes` (default 1 MiB) is a power of two from
+ 32 bytes through 128 MiB.
+- An interpreted NDV is absent or a positive Java long no greater than `Long.MAX_VALUE / 8`.
+- Each interpreted FPP is finite, strictly between 0 and 1, and representable at the required
+ native filter sizes.
+- With a 32-byte cap and interpreted explicit NDV, the NDV/FPP request fits in 32 bytes, because
+ Java ignores that boundary as a maximum for larger requests.
+- Bloom-filter columns use physical Parquet paths supported by the native writer.
+- Bloom-filter columns are not FLOAT or DOUBLE, including nested leaves. Floating-point columns
+ without Bloom filters remain eligible for native writes.
+
+The default Spark profiles have the following Bloom-write eligibility. The loaded Iceberg
+version determines eligibility; Spark's version alone does not enable this feature.
+
+| Spark profile | Pinned Iceberg | Bloom opt-in behavior |
+| ------------- | -------------- | --------------------------------------------- |
+| 3.4 | 1.5.2 | Java writer, even with the opt-in enabled |
+| 3.5 | 1.8.1 | Java writer, even with the opt-in enabled |
+| 4.0 | 1.10.0 | Native eligible; FPP interpreted, NDV ignored |
+| 4.1 | 1.11.0 | Native eligible; FPP and NDV interpreted |
+
+The cap range applies whenever a `write.parquet.bloom-filter-enabled.column.*` property is
+present, even if set to `false`. With no such properties, the cap is unused: unset or any
+positive Java int, such as `100`, keeps an otherwise eligible write native. Malformed, zero,
+or negative caps select Java, which rejects them through its normal validation.
+
+The Bloom option defaults to `false`, selecting Java's writer for Bloom-enabled tables. Enabling
+it accepts byte differences and the pruning tradeoff below; it cannot turn folding off inside
+the pinned parquet-rs writer.
+Enabling it is generally safe when the requested FPP meets your pruning needs: folding preserves
+inserted-value membership and introduces no false negatives. Unsupported settings select Java's
+writer, where normal property validation applies. FLOAT/DOUBLE Bloom writes use Java until native
+writes handle NaN equality consistently with Spark. The fallback is decided from the schema and
+effective write properties before execution, and its reason identifies the affected columns.
+
+Pruning can regress when existing Java-written filters are oversized for their actual values
+and therefore have a much lower FPP than requested. For example, with no explicit NDV, a 64-MiB
+maximum and only 256 distinct values, Java can retain a 64-MiB filter with extra pruning accuracy.
+Rust can shrink that filter toward the target FPP and produce more false positives. Leave the
+option off if your workload relies on that extra accuracy. This can happen with an overestimated
+NDV, skew, or a partly filled row group. With explicit NDV, `max-bytes` is a ceiling; a large cap
+alone does not enlarge the allocation.
+
+Byte parity also requires matching inserted values, physical encodings, and initial sizes.
+The Bloom opt-in also accepts boolean-filter differences: parquet-java writes an all-zero
+filter because it does not insert boolean values, while parquet-rs writes a populated filter.
+Dictionary-only chunks can have a native filter where Java omits it. Comet follows the loaded
+runtime's properties: Iceberg 1.10 ignores NDV, while 1.11 interprets it.
#### Physical file layout only (cosmetic)
diff --git a/native/core/src/execution/operators/iceberg_write.rs b/native/core/src/execution/operators/iceberg_write.rs
index c949b95cf66..604cf68f4cd 100644
--- a/native/core/src/execution/operators/iceberg_write.rs
+++ b/native/core/src/execution/operators/iceberg_write.rs
@@ -78,8 +78,11 @@ use iceberg::ErrorKind;
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel};
use parquet::file::properties::{EnabledStatistics, WriterProperties};
+use parquet::schema::types::ColumnPath;
use serde::{Deserialize, Serialize};
+#[cfg(test)]
+use datafusion_comet_proto::spark_operator::IcebergColumnBloomFilterProps;
use datafusion_comet_proto::spark_operator::{
CompressionCodec as ProtoCompressionCodec, IcebergParquetWriteSettings, IcebergWrite,
IcebergWriteCommon, IcebergWriterMode as ProtoIcebergWriterMode,
@@ -2299,7 +2302,7 @@ fn build_output_batch(
/// the JVM re-derives from the footer with Iceberg's own `MetricsConfig` logic before commit.
fn build_writer_properties(settings: &IcebergParquetWriteSettings) -> DFResult {
let compression = compression_from_proto(settings.compression, settings.compression_level)?;
- Ok(WriterProperties::builder()
+ let mut builder = WriterProperties::builder()
.set_compression(compression)
.set_created_by(settings.created_by.clone())
.set_max_row_group_bytes(Some(settings.row_group_size_bytes as usize))
@@ -2310,8 +2313,147 @@ fn build_writer_properties(settings: &IcebergParquetWriteSettings) -> DFResult ColumnPath {
+ ColumnPath::from(path.split('.').map(str::to_owned).collect::>())
+}
+
+// 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;
+#[cfg(test)]
+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/59.3.0/parquet/src/bloom_filter/mod.rs#L363-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, 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 initial NDV/FPP sizing and power-of-two allocation used by version 59.3.0 of 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. After values are inserted, the 59.3.0 writer folds a
+/// sparse allocation to the smallest power-of-two size that maintains the target FPP:
+/// https://github.com/apache/arrow-rs/blob/59.3.0/parquet/src/bloom_filter/mod.rs#L363-L395
+/// https://github.com/apache/arrow-rs/blob/59.3.0/parquet/src/bloom_filter/mod.rs#L618-L623
+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 initial power-of-two allocation using parquet-rs 59.3.0's public max-NDV/FPP
+/// setters. The writer may subsequently fold that allocation based on the values inserted.
+///
+/// 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 {
+ 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 / 8`, so keep the numeric search explicit.
+ let mut low = 1_u64;
+ let mut high = u64::MAX >> 3;
+ 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() || !(fpp > 0.0 && fpp < 1.0) {
+ return Err(DataFusionError::Internal(format!(
+ "Bloom filter FPP must be finite and strictly between 0 and 1, got {fpp}"
+ )));
+ }
+ if !(BLOOM_FILTER_MIN_BYTES..=BLOOM_FILTER_MAX_BYTES).contains(&bytes)
+ || !bytes.is_power_of_two()
+ {
+ return Err(DataFusionError::Internal(format!(
+ "Bloom filter byte size must be a power of two in [{BLOOM_FILTER_MIN_BYTES}, {BLOOM_FILTER_MAX_BYTES}], got {bytes}"
+ )));
+ }
+ Ok(())
}
fn compression_from_proto(codec: i32, level: Option) -> DFResult {
@@ -2472,6 +2614,17 @@ mod tests {
}
use datafusion_comet_proto::spark_operator::CompressionCodec as ProtoCodec;
+ // Keep this table in sync with the cases in IcebergWriteProtoTranslationSuite. These fixed
+ // expectations make JVM planning and native task sizing fail independently if either drifts.
+ const BLOOM_FILTER_SIZING_CASES: &[(Option, f64, usize, usize)] = &[
+ (None, 0.01, 1024 * 1024, 1024 * 1024),
+ (Some(1), 0.01, 1024 * 1024, 32),
+ (Some(1_000), 0.01, 1024 * 1024, 2 * 1024),
+ (Some(1_000), 0.005, 1024 * 1024, 2 * 1024),
+ (Some(1_000_000), 0.0001, 64, 64),
+ (Some(100_000_000), 0.01, 4 * 1024, 4 * 1024),
+ ];
+
fn base_settings() -> IcebergParquetWriteSettings {
IcebergParquetWriteSettings {
compression: ProtoCodec::Zstd as i32,
@@ -2481,6 +2634,8 @@ mod tests {
dict_size_bytes: 2 * 1024 * 1024,
page_row_limit: 20_000,
created_by: "Apache Iceberg (Comet test)".to_string(),
+ bloom_filters: Vec::new(),
+ bloom_filter_max_bytes: ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES as u64,
}
}
@@ -2560,6 +2715,169 @@ mod tests {
assert_eq!(props.created_by(), "Apache Iceberg 1.7.1 (Comet 0.16.0)");
}
+ #[test]
+ fn enables_bloom_filters_only_for_configured_columns() {
+ let mut settings = base_settings();
+ settings.bloom_filters = vec![
+ IcebergColumnBloomFilterProps {
+ parquet_path: "id".to_string(),
+ fpp: ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ ndv: None,
+ },
+ IcebergColumnBloomFilterProps {
+ parquet_path: "tags.list.element".to_string(),
+ fpp: ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ ndv: None,
+ },
+ IcebergColumnBloomFilterProps {
+ parquet_path: "attrs.key_value.value".to_string(),
+ fpp: ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ ndv: None,
+ },
+ ];
+ let props = build_writer_properties(&settings).unwrap();
+ assert!(props
+ .bloom_filter_properties(&parquet_column_path("id"))
+ .is_some());
+ assert!(props
+ .bloom_filter_properties(&parquet_column_path("tags.list.element"))
+ .is_some());
+ assert!(props
+ .bloom_filter_properties(&parquet_column_path("attrs.key_value.value"))
+ .is_some());
+ assert!(props
+ .bloom_filter_properties(&parquet_column_path("other"))
+ .is_none());
+ }
+
+ #[test]
+ fn bloom_filter_defaults_match_iceberg() {
+ // Iceberg's documented write-property defaults:
+ // https://iceberg.apache.org/docs/latest/configuration/#write-properties
+ let mut settings = base_settings();
+ settings.bloom_filters = vec![IcebergColumnBloomFilterProps {
+ parquet_path: "id".to_string(),
+ fpp: ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ ndv: None,
+ }];
+ let props = build_writer_properties(&settings).unwrap();
+ let bloom = props.bloom_filter_properties(&"id".into()).unwrap();
+ assert_eq!(bloom.fpp(), ICEBERG_DEFAULT_BLOOM_FILTER_FPP);
+ assert_eq!(
+ parquet_rs_bloom_filter_bytes(bloom.ndv(), bloom.fpp()),
+ ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES
+ );
+ }
+
+ #[test]
+ fn bloom_sizing_arithmetic_stays_aligned_with_jvm_planning() {
+ for &(ndv, fpp, max_bytes, expected_bytes) in BLOOM_FILTER_SIZING_CASES {
+ assert_eq!(
+ parquet_mr_bloom_filter_bytes(ndv, fpp, max_bytes),
+ expected_bytes,
+ "ndv={ndv:?} fpp={fpp} max_bytes={max_bytes}"
+ );
+ let synthetic_ndv = synthetic_ndv_for_bloom_filter_bytes(expected_bytes, fpp).unwrap();
+ assert_eq!(
+ parquet_rs_bloom_filter_bytes(synthetic_ndv, fpp),
+ expected_bytes,
+ "ndv={ndv:?} fpp={fpp} max_bytes={max_bytes}"
+ );
+ }
+ }
+
+ #[test]
+ fn synthetic_ndv_hits_every_supported_size_away_from_float_boundaries() {
+ for fpp in [0.0001, ICEBERG_DEFAULT_BLOOM_FILTER_FPP, 0.05, 0.5, 0.99] {
+ let mut bytes = BLOOM_FILTER_MIN_BYTES;
+ while bytes <= BLOOM_FILTER_MAX_BYTES {
+ let ndv = synthetic_ndv_for_bloom_filter_bytes(bytes, fpp).unwrap();
+ assert_eq!(parquet_rs_bloom_filter_bytes(ndv, fpp), bytes);
+ bytes *= 2;
+ }
+ }
+ }
+
+ #[test]
+ fn explicit_ndv_controls_requested_size_and_max_only_caps_it() {
+ let small = parquet_mr_bloom_filter_bytes(
+ Some(10),
+ ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ 64 * 1024 * 1024,
+ );
+ assert!(small < 64 * 1024 * 1024);
+
+ let capped = parquet_mr_bloom_filter_bytes(
+ Some(100_000_000),
+ ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES,
+ );
+ assert_eq!(capped, ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES);
+ let uncapped =
+ parquet_mr_bloom_filter_bytes(None, ICEBERG_DEFAULT_BLOOM_FILTER_FPP, 64 * 1024 * 1024);
+ assert_eq!(uncapped, 64 * 1024 * 1024);
+ }
+
+ #[test]
+ fn minimum_bloom_filter_size_is_representable() {
+ assert_eq!(
+ parquet_mr_bloom_filter_bytes(
+ Some(1),
+ ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ BLOOM_FILTER_MIN_BYTES,
+ ),
+ BLOOM_FILTER_MIN_BYTES
+ );
+ let ndv = synthetic_ndv_for_bloom_filter_bytes(
+ BLOOM_FILTER_MIN_BYTES,
+ ICEBERG_DEFAULT_BLOOM_FILTER_FPP,
+ )
+ .unwrap();
+ assert_eq!(
+ parquet_rs_bloom_filter_bytes(ndv, ICEBERG_DEFAULT_BLOOM_FILTER_FPP),
+ BLOOM_FILTER_MIN_BYTES
+ );
+ }
+
+ #[test]
+ fn impossible_synthetic_ndv_is_rejected() {
+ let err = synthetic_ndv_for_bloom_filter_bytes(
+ ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES,
+ f64::MIN_POSITIVE,
+ )
+ .unwrap_err();
+ assert!(format!("{err}").contains("cannot represent"));
+ }
+
+ #[test]
+ fn binary_search_finds_representable_ndv_when_inverse_candidate_misses() {
+ const FPP_WITH_ROUNDING_SENSITIVE_INVERSE: f64 = f64::from_bits(0x3c6e_f871_5805_0dee);
+ const TARGET_BYTES: usize = 256;
+
+ let denominator = bloom_filter_fpp_denominator(FPP_WITH_ROUNDING_SENSITIVE_INVERSE);
+ let inverse_candidate =
+ ((TARGET_BYTES as f64 * 3.0 / 4.0 * denominator).round() as u64).max(1);
+ assert_ne!(
+ parquet_rs_bloom_filter_bytes(inverse_candidate, FPP_WITH_ROUNDING_SENSITIVE_INVERSE),
+ TARGET_BYTES
+ );
+
+ let searched =
+ synthetic_ndv_for_bloom_filter_bytes(TARGET_BYTES, FPP_WITH_ROUNDING_SENSITIVE_INVERSE)
+ .unwrap();
+ assert_eq!(
+ parquet_rs_bloom_filter_bytes(searched, FPP_WITH_ROUNDING_SENSITIVE_INVERSE),
+ TARGET_BYTES
+ );
+ }
+
+ #[test]
+ fn rejects_zero_bloom_filter_fpp() {
+ let err =
+ validate_bloom_filter_inputs(0.0, ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES).unwrap_err();
+ assert!(format!("{err}").contains("strictly between 0 and 1"));
+ }
+
#[test]
fn rejects_unknown_codec() {
let mut settings = base_settings();
@@ -3035,6 +3353,8 @@ mod tests {
dict_size_bytes: 2 * 1024 * 1024,
page_row_limit: 20_000,
created_by: "Apache Iceberg (Comet integration test)".to_string(),
+ bloom_filters: Vec::new(),
+ bloom_filter_max_bytes: ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES as u64,
};
Arc::new(IcebergWriteCommon {
catalog_properties: HashMap::new(),
diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index 8e720a8897c..fab7d6d01a3 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -782,6 +782,14 @@ enum IcebergWriterMode {
ICEBERG_WRITER_CLUSTERED = 2;
}
+message IcebergColumnBloomFilterProps {
+ string parquet_path = 1;
+ // Effective FPP, with the Iceberg default resolved on the JVM.
+ double fpp = 2;
+ // Absent means allocate the full max-bytes value initially.
+ optional uint64 ndv = 3;
+}
+
// Parquet-rs `WriterProperties` knobs derived from Iceberg table properties on the JVM driver.
// All fields use Iceberg defaults when absent in the table properties; the rust side does not
// re-apply defaults.
@@ -807,6 +815,15 @@ message IcebergParquetWriteSettings {
// String written into parquet file metadata. JVM-side default is
// `"Apache Iceberg (Comet)"`.
string created_by = 7;
+ // Per-column Bloom-filter settings. The JVM driver translates Iceberg logical names to the
+ // physical Parquet leaf paths used by list/map encodings and resolves effective defaults.
+ repeated IcebergColumnBloomFilterProps bloom_filters = 8;
+ // Iceberg `write.parquet.bloom-filter-max-bytes` (default 1 MiB). If any
+ // `write.parquet.bloom-filter-enabled.column.*` key is present (even false), native
+ // eligibility admits only powers of two in [32, 128 MiB]. With no such key, an unset cap
+ // or any positive Java int (including 100) is accepted because it does not affect the file.
+ // Malformed or non-positive caps fall back to Java, which rejects them during validation.
+ uint64 bloom_filter_max_bytes = 9;
}
// Broadcast payload -- one of these per write, identical for every task.
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 4f6e14bebf1..fa2278e16e2 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -144,6 +144,20 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.write.iceberg.bloomFilterFolding.enabled")
+ .category(CATEGORY_EXEC)
+ .doc("Whether Bloom-filter-enabled Iceberg writes may use Comet's native writer. The " +
+ "Apache Arrow Rust parquet writer folds sparse Bloom filters after insertion, which " +
+ "can reduce serialized size while raising realized FPP toward the configured target. " +
+ "When false, these writes use Iceberg's Java writer. This option selects the writer; " +
+ "it cannot disable folding inside parquet-rs. The opt-in also accepts boolean filter " +
+ "differences: Java writes all-zero filters while Rust populates them. " +
+ "Requires spark.comet.write.iceberg.enabled and an identifiable Iceberg version >= 1.10. " +
+ "FLOAT/DOUBLE Bloom columns use Java to preserve NaN equality. Off by default.")
+ .booleanConf
+ .createWithDefault(false)
+
val COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT: ConfigEntry[Int] =
conf("spark.comet.scan.icebergNative.dataFileConcurrencyLimit")
.category(CATEGORY_SCAN)
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 33a0593568d..1c4f8df77a8 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -39,6 +39,13 @@ import org.apache.comet.util.ClassLoaders
*/
object IcebergReflection extends Logging {
+ case class ParquetPathResolution(
+ pathByIcebergColumnName: Map[String, String],
+ pathByFieldId: Map[Int, String],
+ physicalPaths: Set[String],
+ renamedPhysicalPaths: Set[String],
+ floatingPointPhysicalPaths: Set[String])
+
/**
* Iceberg class names used throughout Comet.
*/
@@ -58,6 +65,7 @@ object IcebergReflection extends Logging {
val SPARK_BATCH_QUERY_SCAN = "org.apache.iceberg.spark.source.SparkBatchQueryScan"
val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan"
val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil"
+ val PARQUET_SCHEMA_UTIL = "org.apache.iceberg.parquet.ParquetSchemaUtil"
val TABLE = "org.apache.iceberg.Table"
val DEFAULT_LOCATION_PROVIDER =
"org.apache.iceberg.LocationProviders$DefaultLocationProvider"
@@ -687,6 +695,86 @@ object IcebergReflection extends Logging {
}
}
+ // scalastyle:off line.size.limit
+ /**
+ * Maps Iceberg's logical primitive-column names to their physical Parquet leaf paths.
+ *
+ * This mirrors the map Iceberg Java builds before applying per-column writer settings. In
+ * particular, Parquet's canonical three-level encodings insert `list` for array elements and
+ * `key_value` for map keys/values, so logical names such as `tags.element` and `attrs.value`
+ * cannot be passed directly to Parquet writer properties.
+ *
+ * See:
+ * https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java#L411-L421
+ */
+ // scalastyle:on line.size.limit
+ def getParquetPathResolution(schema: Any): Option[ParquetPathResolution] = {
+ import scala.jdk.CollectionConverters._
+ try {
+ val parquetSchemaUtil = loadClass(ClassNames.PARQUET_SCHEMA_UTIL)
+ val parquetSchema = parquetSchemaUtil
+ .getMethod("convert", loadClass(ClassNames.SCHEMA), classOf[String])
+ .invoke(null, schema.asInstanceOf[AnyRef], "table")
+ val columns = getMethod(parquetSchema.getClass, "getColumns")
+ .invoke(parquetSchema)
+ .asInstanceOf[java.util.List[AnyRef]]
+ val findColumnName = getMethod(schema.getClass, "findColumnName", classOf[Int])
+ val findField = getMethod(schema.getClass, "findField", classOf[Int])
+
+ val resolved = columns.asScala.flatMap { column =>
+ val primitiveType = getMethod(column.getClass, "getPrimitiveType").invoke(column)
+ val floatingPoint = Set("FLOAT", "DOUBLE").contains(
+ getMethod(primitiveType.getClass, "getPrimitiveTypeName")
+ .invoke(primitiveType)
+ .toString)
+ val parquetId = getMethod(primitiveType.getClass, "getId").invoke(primitiveType)
+ Option(parquetId).flatMap { id =>
+ val fieldId = getMethod(id.getClass, "intValue").invoke(id).asInstanceOf[Int]
+ Option(findColumnName.invoke(schema, Int.box(fieldId))).map { icebergName =>
+ val parquetPathParts = getMethod(column.getClass, "getPath")
+ .invoke(column)
+ .asInstanceOf[Array[String]]
+ var parquetType = parquetSchema.asInstanceOf[AnyRef]
+ val renamed = parquetPathParts.exists { pathPart =>
+ parquetType = getMethod(parquetType.getClass, "getType", classOf[String])
+ .invoke(parquetType, pathPart)
+ .asInstanceOf[AnyRef]
+ Option(getMethod(parquetType.getClass, "getId").invoke(parquetType)).exists {
+ parquetFieldId =>
+ val idValue = getMethod(parquetFieldId.getClass, "intValue")
+ .invoke(parquetFieldId)
+ .asInstanceOf[Int]
+ Option(findField.invoke(schema, Int.box(idValue))).exists { icebergField =>
+ val icebergFieldName = getMethod(icebergField.getClass, "name")
+ .invoke(icebergField)
+ .asInstanceOf[String]
+ icebergFieldName != pathPart
+ }
+ }
+ }
+ (
+ fieldId,
+ icebergName.asInstanceOf[String],
+ parquetPathParts.mkString("."),
+ renamed,
+ floatingPoint)
+ }
+ }
+ }.toSeq
+ Some(
+ ParquetPathResolution(
+ resolved.map { case (_, name, path, _, _) => name -> path }.toMap,
+ resolved.map { case (fieldId, _, path, _, _) => fieldId -> path }.toMap,
+ resolved.map { case (_, _, path, _, _) => path }.toSet,
+ resolved.collect { case (_, _, path, true, _) => path }.toSet,
+ resolved.collect { case (_, _, path, _, true) => path }.toSet))
+ } catch {
+ case e: Exception =>
+ logError(s"Iceberg reflection failure: Parquet column paths: ${e.getMessage}")
+ None
+ }
+ }
+
/**
* All schema versions a table has had (table.schemas().values()), for resolving field ids of
* columns that have since been dropped -- mirrors Iceberg-Java's FieldLookup. table.schemas()
@@ -1579,6 +1667,12 @@ object IcebergReflection extends Logging {
def tablePropertyIntConstant(fieldName: String): Int =
readTablePropertiesField(fieldName).asInstanceOf[Integer].intValue()
+ def tablePropertyDoubleConstantOpt(fieldName: String): Option[Double] =
+ tablePropertiesClassOpt.flatMap { cls =>
+ try Some(cls.getField(fieldName).get(null).asInstanceOf[java.lang.Double].doubleValue())
+ catch { case _: NoSuchFieldException => None }
+ }
+
/**
* Like [[tablePropertyConstant]] but returns `None` when the constant is absent in the Iceberg
* version on the classpath rather than throwing. Used to gate behaviour that only some Iceberg
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
index b385668291d..21e5828e435 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
@@ -50,6 +50,10 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
IcebergReflection.tablePropertyConstant("WRITE_LOCATION_PROVIDER_IMPL")
lazy val BloomFilterColumnEnabledPrefix: String =
IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX")
+ lazy val ParquetBloomFilterMaxBytes: String =
+ IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_MAX_BYTES")
+ val ParquetBloomFilterColumnFppPrefix = "write.parquet.bloom-filter-fpp.column."
+ val ParquetBloomFilterColumnNdvPrefix = "write.parquet.bloom-filter-ndv.column."
lazy val ParquetRowGroupCheckMinRecordCount: String =
IcebergReflection.tablePropertyConstant("PARQUET_ROW_GROUP_CHECK_MIN_RECORD_COUNT")
lazy val ParquetRowGroupCheckMinRecordCountDefault: Int =
@@ -281,10 +285,14 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
PropertyKeys.ParquetRowGroupCheckMaxRecordCount,
PropertyKeys.ParquetPageVersion,
PropertyKeys.ParquetShredVariants,
- PropertyKeys.ParquetVariantBufferSize)
+ PropertyKeys.ParquetVariantBufferSize,
+ PropertyKeys.ParquetBloomFilterMaxBytes)
private lazy val vettedParquetWritePrefixes: Seq[String] =
- Seq(PropertyKeys.BloomFilterColumnEnabledPrefix)
+ Seq(
+ PropertyKeys.BloomFilterColumnEnabledPrefix,
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix,
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix)
override def getSupportLevel(op: IcebergWriteExec): SupportLevel =
try {
@@ -328,7 +336,12 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
() => op.session.sessionState.newHadoopConf(),
sessionConf => effectiveHadoopConf(op, table, sessionConf))
val context =
- TriggerContext(table, tableProperties ++ writeProperties, sparkWrite, hadoopConfs)
+ TriggerContext(
+ table,
+ tableProperties ++ writeProperties,
+ sparkWrite,
+ hadoopConfs,
+ CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.get(op.conf))
triggers.iterator.map(rule => rule(context)).collectFirst { case Some(reason) => reason }
}
@@ -336,7 +349,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
table: Any,
properties: Map[String, String],
sparkWrite: Any,
- hadoopConfs: LazyHadoopConfs) {
+ hadoopConfs: LazyHadoopConfs,
+ bloomFilterFoldingEnabled: Boolean) {
def hadoopConf: Configuration = hadoopConfs.session
def s3HadoopConf: Configuration = hadoopConfs.effective
}
@@ -417,12 +431,15 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
requireSupportedColumnTypes,
requireNoVoidFieldWithDroppedSource,
requireNoEncryptionPrefix,
- requireNoBloomFilterColumnsEnabled,
requireRowGroupCheckMinRecordCountAtDefault,
requireRowGroupCheckMaxRecordCountAtDefault,
requireParquetPageVersionDefault,
requireShredVariantsDisabled,
requireNativeSupportedCompressionLevel,
+ requireSupportedBloomFilterRuntime,
+ requireNativeSupportedBloomFilterProperties,
+ requireBloomFilterFoldingOptIn,
+ requireNoFloatingPointBloomFilters,
requireOnlyVettedParquetWriteProperties,
requirePropertyAbsent(
PropertyKeys.ParquetEnableDictionary,
@@ -563,13 +580,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
// `CometIcebergWriteExec`), so every `write.metadata.metrics.*` value behaves exactly as it
// does on the iceberg-java path.
- private val requireNoBloomFilterColumnsEnabled: TriggerRule = ctx => {
- val prefix = PropertyKeys.BloomFilterColumnEnabledPrefix
- ctx.properties
- .find { case (k, v) => k.startsWith(prefix) && v.equalsIgnoreCase("true") }
- .map { case (k, _) => s"$k=true: bloom filters unsupported" }
- }
-
private val requireParquetPageVersionDefault: TriggerRule = ctx => {
val key = PropertyKeys.ParquetPageVersion
ctx.properties
@@ -594,6 +604,268 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
private val requireNativeSupportedCompressionLevel: TriggerRule = ctx =>
IcebergWriteProtoTranslation.compressionLevelRejection(ctx.properties)
+ // These are Apache Parquet Java BlockSplitBloomFilter implementation bounds, not Iceberg
+ // TableProperties constants, so they cannot be obtained through IcebergReflection:
+ // scalastyle:off line.size.limit
+ // https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L40-L50
+ // scalastyle:on line.size.limit
+ private[operator] val MinBloomFilterBytes = 32
+ private[operator] val MaxBloomFilterBytes = 128 * 1024 * 1024
+ private val BloomFilterHashProbes = 8
+ private val MaxNonOverflowingBloomFilterNdv = Long.MaxValue / BloomFilterHashProbes
+
+ /**
+ * Keep only Bloom shape properties interpreted by the Iceberg runtime on the classpath. Older
+ * Iceberg releases leave these table properties untouched but do not pass them to parquet-mr.
+ * Ignoring them here preserves that version's JVM-writer behavior while allowing the remaining
+ * supported Bloom configuration to execute natively.
+ */
+ private def interpretedBloomFilterProperties(
+ properties: Map[String, String]): Map[String, String] = {
+ val unsupportedPrefixes = Seq(
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix ->
+ "PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX",
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix ->
+ "PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX").collect {
+ case (prefix, constant) if IcebergReflection.tablePropertyConstantOpt(constant).isEmpty =>
+ prefix
+ }
+ properties.filterNot { case (key, _) => unsupportedPrefixes.exists(key.startsWith) }
+ }
+
+ /**
+ * parquet-rs 59.3.0 initially represents Bloom filters as a power-of-two number of bytes and
+ * may fold that allocation after values are inserted. parquet-mr accepts arbitrary caps and,
+ * when one binds, serializes that exact length. Keep those writes on the classic path instead
+ * of silently changing the number of usable Bloom blocks.
+ */
+ private val requireNativeSupportedBloomFilterProperties: TriggerRule = ctx => {
+ val properties = interpretedBloomFilterProperties(ctx.properties)
+ val configuredMaxBytes = properties.get(PropertyKeys.ParquetBloomFilterMaxBytes).map { raw =>
+ raw -> scala.util.Try(java.lang.Integer.parseInt(raw)).toOption
+ }
+ val malformedMaxRejection = configuredMaxBytes.collect {
+ case (raw, None) =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw is not a Java int"
+ case (raw, Some(value)) if value <= 0 =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw must be positive"
+ }
+ // Iceberg visits every enabled-prefix entry and applies enabled, FPP, then NDV. With no such
+ // entry, a positive, parseable maximum cannot affect the file and must not force an
+ // otherwise-compatible write onto the classic path. Keep rejecting malformed and
+ // non-positive values because Iceberg validates them even with no Bloom column.
+ val configured = properties.iterator.collect {
+ case (key, _) if key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) =>
+ key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length)
+ }.toSeq
+ if (configured.isEmpty) {
+ malformedMaxRejection
+ } else {
+ val maxRejection = malformedMaxRejection.orElse {
+ configuredMaxBytes.collect {
+ case (_, Some(value))
+ if value < MinBloomFilterBytes || value > MaxBloomFilterBytes ||
+ (value & (value - 1)) != 0 =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$value must be a power of two " +
+ s"in [$MinBloomFilterBytes, $MaxBloomFilterBytes] for native writes"
+ }
+ }
+
+ maxRejection.orElse {
+ val maxBytes = configuredMaxBytes
+ .flatMap(_._2)
+ .getOrElse(IcebergWriteProtoTranslation.Defaults.BloomFilterMaxBytes)
+ // Validate associated shape properties even for enabled=false; a valid NDV re-enables
+ // the filter in parquet-mr.
+ configured.iterator
+ .flatMap { column =>
+ val fppKey = PropertyKeys.ParquetBloomFilterColumnFppPrefix + column
+ val ndvKey = PropertyKeys.ParquetBloomFilterColumnNdvPrefix + column
+ val parsedFpp = properties.get(fppKey) match {
+ case Some(raw) => scala.util.Try(java.lang.Double.parseDouble(raw)).toOption
+ case None => Some(IcebergWriteProtoTranslation.Defaults.BloomFilterFpp)
+ }
+ val parsedNdv = properties
+ .get(ndvKey)
+ .flatMap(raw => scala.util.Try(java.lang.Long.parseLong(raw)).toOption)
+ val fppError = properties.get(fppKey).flatMap { raw =>
+ parsedFpp match {
+ case Some(value)
+ if value > 0.0d && value < 1.0d && java.lang.Double.isFinite(value) &&
+ bloomFilterSizesRepresentable(maxBytes, value) =>
+ None
+ case Some(value)
+ if value > 0.0d && value < 1.0d && java.lang.Double.isFinite(value) =>
+ Some(s"$fppKey=$raw cannot represent the configured native Bloom sizes")
+ case _ => Some(s"$fppKey=$raw must be a finite double strictly between 0 and 1")
+ }
+ }
+ val ndvError = properties.get(ndvKey).flatMap { raw =>
+ parsedNdv match {
+ case Some(value) if value > 0L && value <= MaxNonOverflowingBloomFilterNdv => None
+ case Some(value) if value > MaxNonOverflowingBloomFilterNdv =>
+ Some(s"$ndvKey=$raw exceeds $MaxNonOverflowingBloomFilterNdv; " +
+ "parquet-mr Bloom sizing may overflow")
+ case _ => Some(s"$ndvKey=$raw must be a positive Java long")
+ }
+ }
+ val ignoredMinimumCapError = parsedNdv.collect {
+ case ndv
+ if maxBytes == MinBloomFilterBytes && ndv > 0L &&
+ ndv <= MaxNonOverflowingBloomFilterNdv &&
+ parsedFpp.exists(
+ parquetMrRequestedBloomFilterBytes(ndv, _) > MinBloomFilterBytes) =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$MinBloomFilterBytes is ignored by " +
+ s"parquet-mr for $ndvKey=$ndv"
+ }
+ Seq(fppError, ndvError, ignoredMinimumCapError).flatten
+ }
+ .toSeq
+ .headOption
+ }
+ }
+ }
+
+ private val requireSupportedBloomFilterRuntime: TriggerRule = ctx => {
+ val prefixes = Seq(
+ PropertyKeys.BloomFilterColumnEnabledPrefix,
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix,
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix)
+ if (ctx.properties.keys.exists(key => prefixes.exists(key.startsWith))) {
+ bloomFilterRuntimeRejection(IcebergReflection.icebergVersion())
+ } else {
+ None
+ }
+ }
+
+ private[operator] def bloomFilterRuntimeRejection(icebergVersion: String): Option[String] = {
+ val version =
+ if (Option(icebergVersion).exists(_.matches("\\d+\\.\\d+\\.\\d+(?:[-+].+)?"))) {
+ val parts = icebergVersion.split("[.-]", 3)
+ scala.util.Try((parts(0).toInt, parts(1).toInt)).toOption
+ } else {
+ None
+ }
+ version match {
+ case Some((1, minor)) if minor >= 10 => None
+ case Some((1, _)) =>
+ Some(s"Native Bloom writes require Iceberg 1.10 or later; found '$icebergVersion'")
+ case _ =>
+ Some(
+ s"Cannot determine Iceberg Bloom column-name convention for version '$icebergVersion'")
+ }
+ }
+
+ private[operator] def bloomFilterParquetPaths(
+ configuredColumns: Seq[String],
+ resolution: IcebergReflection.ParquetPathResolution,
+ icebergVersion: String = IcebergReflection.icebergVersion()): Map[String, String] = {
+ bloomFilterRuntimeRejection(icebergVersion).foreach { reason =>
+ throw new IllegalArgumentException(reason)
+ }
+ resolution.pathByIcebergColumnName.filter { case (column, _) =>
+ configuredColumns.contains(column)
+ }
+ }
+
+ // A floating-point Bloom filter can reject a SQL-equal NaN with a different bit payload.
+ // Until native writes canonicalize these values, preserve Java's writer for affected leaves.
+ private val requireNoFloatingPointBloomFilters: TriggerRule = ctx => {
+ val properties = interpretedBloomFilterProperties(ctx.properties)
+ val columns = IcebergWriteProtoTranslation.enabledBloomFilterColumnNames(properties)
+ if (columns.isEmpty) {
+ None
+ } else {
+ val schema = IcebergReflection
+ .getWriteSchemaFromSparkWrite(ctx.sparkWrite)
+ .getOrElse(
+ throw new IllegalStateException("could not resolve schema for Bloom type checking"))
+ val resolution = IcebergReflection
+ .getParquetPathResolution(schema)
+ .getOrElse(
+ throw new IllegalStateException(
+ "could not resolve Parquet paths for Bloom type checking"))
+ val paths = bloomFilterParquetPaths(columns, resolution)
+ val affected = columns.filter { column =>
+ paths.get(column).exists(resolution.floatingPointPhysicalPaths.contains)
+ }
+ if (affected.isEmpty) {
+ None
+ } else {
+ Some(
+ s"Bloom filters on FLOAT/DOUBLE columns ${affected.sorted.mkString(", ")} " +
+ "require the Java writer to preserve NaN equality")
+ }
+ }
+ }
+
+ private val requireBloomFilterFoldingOptIn: TriggerRule = ctx => {
+ val properties = interpretedBloomFilterProperties(ctx.properties)
+ if (IcebergWriteProtoTranslation.hasEnabledBloomFilters(properties) &&
+ !ctx.bloomFilterFoldingEnabled) {
+ Some(
+ s"${CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key}=false " +
+ "(native Parquet Bloom-filter folding is not enabled)")
+ } else {
+ None
+ }
+ }
+
+ // Planning-time counterpart of the native inverse-NDV check. A target B is safely encoded by
+ // aiming at 3B/4, in the interior of parquet-rs's (B/2, B] round-up interval. Requiring every
+ // power-of-two through the configured cap is conservative and keeps pathological-but-valid
+ // floating-point FPPs on the JVM path rather than discovering them after task launch.
+ private[operator] def bloomFilterSizesRepresentable(maxBytes: Int, fpp: Double): Boolean = {
+ val denominator =
+ -Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble))
+ if (!java.lang.Double.isFinite(denominator) || denominator <= 0.0d) return false
+
+ Iterator
+ .iterate(MinBloomFilterBytes)(_ * 2)
+ .takeWhile(_ <= maxBytes)
+ .forall { target =>
+ val ndv = Math.max(1L, Math.round(target.toDouble * 0.75d * denominator))
+ val calculatedBits = (-BloomFilterHashProbes.toDouble * ndv.toDouble /
+ Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble))).toLong
+ val rawBytes = Math.max(
+ MinBloomFilterBytes.toLong,
+ Math.min(MaxBloomFilterBytes.toLong, calculatedBits / 8L))
+ val allocated = java.lang.Long.highestOneBit(rawBytes - 1L) << 1
+ allocated == target.toLong
+ }
+ }
+
+ /** The uncapped byte count parquet-mr passes to its explicit-NDV constructor path. */
+ private[operator] def parquetMrRequestedBloomFilterBytes(ndv: Long, fpp: Double): Int = {
+ // Keep the long multiplication before floating-point conversion to match parquet-mr:
+ // scalastyle:off line.size.limit
+ // https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L277-L301
+ // scalastyle:on line.size.limit
+ val calculated = (-BloomFilterHashProbes.toLong * ndv).toDouble /
+ Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble))
+ val maxBits = MaxBloomFilterBytes << 3
+ var bits = calculated.toInt
+ if (bits > maxBits || calculated < 0.0d) bits = maxBits
+ val bitsPerBlock = MinBloomFilterBytes << 3
+ bits = (bits + bitsPerBlock - 1) & ~bitsPerBlock
+ Math.max(bits, bitsPerBlock) / 8
+ }
+
+ /** The power-of-two byte allocation Comet asks parquet-rs to make for parquet-mr parity. */
+ private[operator] def parquetMrBloomFilterBytes(
+ ndv: Option[Long],
+ fpp: Double,
+ maxBytes: Int): Int = {
+ ndv match {
+ case None => maxBytes
+ case Some(value) =>
+ val requested = parquetMrRequestedBloomFilterBytes(value, fpp)
+ val bounded = Math.max(MinBloomFilterBytes, Math.min(MaxBloomFilterBytes, requested))
+ val allocated = java.lang.Integer.highestOneBit(bounded - 1) << 1
+ Math.min(allocated, maxBytes)
+ }
+ }
+
private val requireOnlyVettedParquetWriteProperties: TriggerRule = ctx =>
ctx.properties
.find { case (k, _) =>
@@ -1120,9 +1392,39 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
// `option("write-parquet-compression-codec", "gzip")`) survive into the native writer.
val resolvedWriteProperties =
IcebergReflection.getWritePropertiesFromSparkWrite(sparkWrite).getOrElse(Map.empty)
- val effectiveProperties = properties ++ resolvedWriteProperties
- val parquetSettings =
- IcebergWriteProtoTranslation.buildParquetSettings(effectiveProperties, createdBy)
+ val effectiveProperties =
+ interpretedBloomFilterProperties(properties ++ resolvedWriteProperties)
+ val parquetPathByIcebergColumnName =
+ if (IcebergWriteProtoTranslation.hasEnabledBloomFilters(effectiveProperties)) {
+ val resolution = IcebergReflection.getParquetPathResolution(writeSchema).getOrElse {
+ withFallbackReason(op, "Could not resolve physical Parquet paths for Bloom filters")
+ return None
+ }
+ val configuredColumns =
+ IcebergWriteProtoTranslation.enabledBloomFilterColumnNames(effectiveProperties)
+ val resolvedPaths = bloomFilterParquetPaths(configuredColumns, resolution)
+ val renamedColumns = resolvedPaths.collect {
+ case (configuredName, path) if resolution.renamedPhysicalPaths.contains(path) =>
+ configuredName
+ }
+ if (renamedColumns.nonEmpty) {
+ // Iceberg Java sanitizes these Parquet names, while the pinned iceberg-rust Arrow
+ // conversion preserves them. Passing the Java path to parquet-rs would silently miss
+ // the native writer column, so retain the JVM writer until paths travel structurally.
+ withFallbackReason(
+ op,
+ "Bloom-filter columns are renamed in Iceberg Java's Parquet schema: " +
+ renamedColumns.mkString(", "))
+ return None
+ }
+ resolvedPaths
+ } else {
+ Map.empty[String, String]
+ }
+ val parquetSettings = IcebergWriteProtoTranslation.buildParquetSettings(
+ effectiveProperties,
+ createdBy,
+ parquetPathByIcebergColumnName)
// `FileIO.properties()` misses configuration a HadoopFileIO carries through the Hadoop
// Configuration instead (fs.s3a.* credentials, custom endpoint, path-style access), which
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala b/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala
index 531f3c227dd..c51778617bd 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala
@@ -57,6 +57,15 @@ object IcebergWriteProtoTranslation {
IcebergReflection.tablePropertyConstant("PARQUET_PAGE_ROW_LIMIT")
lazy val ParquetDictSizeBytes: String =
IcebergReflection.tablePropertyConstant("PARQUET_DICT_SIZE_BYTES")
+ lazy val ParquetBloomFilterColumnEnabledPrefix: String =
+ IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX")
+ lazy val ParquetBloomFilterMaxBytes: String =
+ IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_MAX_BYTES")
+ // These were added to Iceberg after bloom enablement/max-bytes. Literals let the translation
+ // support them without a hard binary dependency; the caller removes either prefix when the
+ // Iceberg runtime does not interpret it, matching that runtime's JVM writer.
+ val ParquetBloomFilterColumnFppPrefix = "write.parquet.bloom-filter-fpp.column."
+ val ParquetBloomFilterColumnNdvPrefix = "write.parquet.bloom-filter-ndv.column."
}
/** Iceberg's numeric defaults, pulled at runtime so they stay in lock-step with the runtime. */
@@ -69,6 +78,14 @@ object IcebergWriteProtoTranslation {
IcebergReflection.tablePropertyIntConstant("PARQUET_DICT_SIZE_BYTES_DEFAULT").toLong
lazy val PageRowLimit: Int =
IcebergReflection.tablePropertyIntConstant("PARQUET_PAGE_ROW_LIMIT_DEFAULT")
+ lazy val BloomFilterMaxBytes: Int =
+ IcebergReflection.tablePropertyIntConstant("PARQUET_BLOOM_FILTER_MAX_BYTES_DEFAULT")
+ // Iceberg introduced the public constant together with the FPP property. Keep the literal
+ // fallback for runtimes old enough not to expose it; 0.01 is also parquet-mr's default.
+ lazy val BloomFilterFpp: Double =
+ IcebergReflection
+ .tablePropertyDoubleConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_DEFAULT")
+ .getOrElse(0.01d)
}
/**
@@ -100,10 +117,43 @@ object IcebergWriteProtoTranslation {
}
}
+ private def configuredBloomFilterColumnNames(props: Map[String, String]): Seq[String] =
+ props.iterator
+ .collect {
+ case (key, _) if key.startsWith(Keys.ParquetBloomFilterColumnEnabledPrefix) =>
+ key.substring(Keys.ParquetBloomFilterColumnEnabledPrefix.length)
+ }
+ .toSeq
+ .sorted
+
+ private[operator] def enabledBloomFilterColumnNames(props: Map[String, String]): Seq[String] =
+ configuredBloomFilterColumnNames(props).filter { column =>
+ val enabled =
+ java.lang.Boolean.valueOf(props(Keys.ParquetBloomFilterColumnEnabledPrefix + column))
+ // Iceberg applies enabled, FPP, and NDV in that order. parquet-mr's NDV setter enables the
+ // column, so an explicit NDV wins over enabled=false.
+ enabled || props.contains(Keys.ParquetBloomFilterColumnNdvPrefix + column)
+ }
+
+ def hasEnabledBloomFilters(props: Map[String, String]): Boolean =
+ enabledBloomFilterColumnNames(props).nonEmpty
+
+ /**
+ * Test convenience for schemas where Iceberg logical names and physical Parquet paths are
+ * identical. Production translation must supply Iceberg Java's logical-to-physical path map.
+ */
+ private[operator] def buildParquetSettings(
+ props: Map[String, String],
+ createdBy: String): IcebergParquetWriteSettings = {
+ val identityPaths = enabledBloomFilterColumnNames(props).map(name => name -> name).toMap
+ buildParquetSettings(props, createdBy, identityPaths)
+ }
+
/** Builds the parquet settings message. Pure: no SparkWrite or Iceberg `Table` access. */
def buildParquetSettings(
props: Map[String, String],
- createdBy: String): IcebergParquetWriteSettings = {
+ createdBy: String,
+ parquetPathByIcebergColumnName: Map[String, String]): IcebergParquetWriteSettings = {
val rowGroupSize =
parseJavaInt(props, Keys.ParquetRowGroupSizeBytes, Defaults.RowGroupSizeBytes.toInt).toLong
val pageSize =
@@ -112,6 +162,29 @@ object IcebergWriteProtoTranslation {
parseJavaInt(props, Keys.ParquetDictSizeBytes, Defaults.DictSizeBytes.toInt).toLong
val pageRowLimit = parseJavaInt(props, Keys.ParquetPageRowLimit, Defaults.PageRowLimit)
val compression = resolveCompression(props)
+ // Iceberg properties use logical schema paths, while Parquet writer properties require the
+ // physical leaf path. Missing fields are skipped, matching Iceberg Java's writer behavior.
+ val bloomFilterColumns = enabledBloomFilterColumnNames(props)
+ .flatMap { icebergName =>
+ parquetPathByIcebergColumnName.get(icebergName).map(icebergName -> _)
+ }
+ .sortBy(_._2)
+ val bloomFilterMaxBytes =
+ parseJavaInt(props, Keys.ParquetBloomFilterMaxBytes, Defaults.BloomFilterMaxBytes).toLong
+ val bloomFilters = bloomFilterColumns.map { case (icebergName, parquetPath) =>
+ val builder = IcebergColumnBloomFilterProps
+ .newBuilder()
+ .setParquetPath(parquetPath)
+ .setFpp(
+ props
+ .get(Keys.ParquetBloomFilterColumnFppPrefix + icebergName)
+ .map(java.lang.Double.parseDouble)
+ .getOrElse(Defaults.BloomFilterFpp))
+ props
+ .get(Keys.ParquetBloomFilterColumnNdvPrefix + icebergName)
+ .foreach(value => builder.setNdv(java.lang.Long.parseLong(value)))
+ builder.build()
+ }
val builder = IcebergParquetWriteSettings
.newBuilder()
.setCompression(compression)
@@ -120,6 +193,8 @@ object IcebergWriteProtoTranslation {
.setDictSizeBytes(dictSize)
.setPageRowLimit(pageRowLimit)
.setCreatedBy(createdBy)
+ .addAllBloomFilters(bloomFilters.asJava)
+ .setBloomFilterMaxBytes(bloomFilterMaxBytes)
resolveCompressionLevel(props, compression).foreach(builder.setCompressionLevel)
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
index 6f752267a63..a2e40fa9dab 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
@@ -19,7 +19,8 @@
package org.apache.comet
-import java.io.File
+import java.io.{ByteArrayOutputStream, File}
+import java.nio.ByteBuffer
import java.nio.file.{FileAlreadyExistsException, Files}
import java.sql.Timestamp
import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
@@ -38,6 +39,7 @@ import org.json4s.jackson.JsonMethods.parse
import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.apache.parquet.io.api.Binary
import org.apache.spark.{CometListenerBusUtils, SparkConf, SparkException, SparkThrowable, Success, TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
@@ -1241,6 +1243,859 @@ class CometIcebergWriteActionSuite
}
}
+ test("legacy Bloom settings use Java despite the native folding opt-in") {
+ assumeNativeAcceleration()
+ assume(!icebergVersionAtLeast(1, 10), "legacy runtime only")
+ withIcebergCatalog { _ =>
+ Seq("true", "false").zipWithIndex.foreach { case (enabled, index) =>
+ val table = s"legacy_bloom_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-enabled.column.id'='$enabled'"))
+ assertNativeWriteDoesNotEngage(table, Seq(1)) {
+ spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, 'region', 1.0)")
+ }
+ }
+ }
+ }
+
+ test("native acceleration: writes configured Iceberg Parquet bloom filters") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(
+ "native_bloom",
+ partitionSpec = "",
+ properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
+
+ val expectedIds = 0 until 256
+ assertNativeWriteEngages("native_bloom", expectedIds) {
+ spark.sql(
+ "INSERT INTO cat.db.native_bloom " +
+ "SELECT CAST(id AS INT), CONCAT('region-', CAST(id % 4 AS STRING)), " +
+ "CAST(id AS DOUBLE) FROM range(256)")
+ }
+
+ assertParquetBloomFilters("native_bloom", enabledColumns = Set("id"))
+
+ }
+ }
+
+ test("Bloom-filter folding config selects the native or JVM Iceberg writer") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'")
+ createTable("bloom_folding_enabled", partitionSpec = "", properties)
+ createTable("bloom_folding_disabled", partitionSpec = "", properties)
+
+ assertNativeWriteEngages("bloom_folding_enabled", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_folding_enabled VALUES (1, 'region', 1.0)")
+ }
+ assertNativeWriteDoesNotEngage(
+ "bloom_folding_disabled",
+ Seq(1),
+ bloomFilterFoldingEnabled = false) {
+ spark.sql("INSERT INTO cat.db.bloom_folding_disabled VALUES (1, 'region', 1.0)")
+ }
+ assertParquetBloomFilters("bloom_folding_enabled", enabledColumns = Set("id"))
+ assertParquetBloomFilters("bloom_folding_disabled", enabledColumns = Set("id"))
+ }
+ }
+
+ test("floating-point Bloom writes fall back and preserve NaN predicate results") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ val table = "floating_bloom_nan"
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT, region STRING, amount DOUBLE, f FLOAT
+ ) USING iceberg TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.amount'='true',
+ 'write.parquet.bloom-filter-enabled.column.f'='true',
+ 'write.parquet.dict-size-bytes'='1'
+ )
+ """)
+ val count = 4096
+ assertNativeWriteDoesNotEngage(table, 0 until count) {
+ spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table
+ SELECT CAST(id AS INT), 'region', CAST('NaN' AS DOUBLE), CAST('NaN' AS FLOAT)
+ FROM range(0, $count, 1, 1)
+ """)
+ }
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$table.data_files")
+ .collect()
+ .map(_.getString(0))
+ assert(paths.nonEmpty)
+ val conf = spark.sparkContext.hadoopConfiguration
+ val bloomKey = "parquet.filter.bloom.enabled"
+ val previous = Option(conf.get(bloomKey))
+ try {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ Seq(false, true).foreach { enabled =>
+ conf.setBoolean(bloomKey, enabled)
+ val written = spark.read.parquet(paths: _*)
+ assert(written.filter("amount = CAST('NaN' AS DOUBLE)").count() == count)
+ assert(written.filter("f = CAST('NaN' AS FLOAT)").count() == count)
+ }
+ }
+ } finally {
+ previous match {
+ case Some(value) => conf.set(bloomKey, value)
+ case None => conf.unset(bloomKey)
+ }
+ }
+ }
+ }
+
+ test("floating-point columns without Bloom filters keep native writes") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ createTable(
+ "unfiltered_floating_column",
+ partitionSpec = "",
+ properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
+ assertNativeWriteEngages("unfiltered_floating_column", Seq(1)) {
+ spark.sql(
+ s"INSERT INTO $catalog.$ns.unfiltered_floating_column " +
+ "VALUES (1, 'region', CAST('NaN' AS DOUBLE))")
+ }
+ assertParquetBloomFilters("unfiltered_floating_column", enabledColumns = Set("id"))
+ }
+ }
+
+ test("enabled=false plus NDV matches JVM behavior for the Iceberg runtime") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val configuredFpp = 0.01
+ val configuredNdv = 1000
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ s"'write.parquet.bloom-filter-fpp.column.id'='$configuredFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'")
+ createTable("bloom_false_ndv_native", partitionSpec = "", properties)
+ createTable("bloom_false_ndv_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) " +
+ s"FROM range(0, $configuredNdv, 1, 1)")
+
+ assertNativeWriteEngages("bloom_false_ndv_native", 0 until configuredNdv) {
+ insert("bloom_false_ndv_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_false_ndv_jvm")
+ }
+
+ val nativeHasBloom = parquetBloomFilterPresent("bloom_false_ndv_native", "id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_false_ndv_jvm", "id")
+ assert(nativeHasBloom == jvmHasBloom)
+ if (icebergSupportsBloomNdv) {
+ assert(nativeHasBloom, "an interpreted NDV must re-enable the Bloom filter")
+ val native = parquetBloomFilterBytes("bloom_false_ndv_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_false_ndv_jvm", "id")
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ assert(!nativeHasBloom, "an uninterpreted NDV must not override enabled=false")
+ }
+ }
+ }
+
+ test("enabled=false preserves JVM validation errors for malformed FPP and NDV") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ Seq(
+ ("fpp", "'write.parquet.bloom-filter-fpp.column.id'='garbage'", icebergSupportsBloomFpp),
+ ("ndv", "'write.parquet.bloom-filter-ndv.column.id'='garbage'", icebergSupportsBloomNdv))
+ .foreach { case (suffix, malformedProperty, interpreted) =>
+ val properties =
+ Some(s"'write.parquet.bloom-filter-enabled.column.id'='false', $malformedProperty")
+ val nativeTable = s"bloom_false_bad_${suffix}_native"
+ val jvmTable = s"bloom_false_bad_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(table, partitionSpec = "", properties = properties)
+ }
+
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table VALUES (1, 'region', 1.0)")
+
+ if (interpreted) {
+ val withComet = intercept[Throwable] {
+ withNativeEnabled(insert(nativeTable))
+ }
+ val withJvm = intercept[Throwable] {
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert(jvmTable)
+ }
+ }
+ val cometCause = exceptionChain(withComet).last
+ val jvmCause = exceptionChain(withJvm).last
+ assert(cometCause.getClass == jvmCause.getClass)
+ assert(cometCause.getMessage == jvmCause.getMessage)
+ } else {
+ assertNativeWriteEngages(nativeTable, Seq(1))(insert(nativeTable))
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert(jvmTable)
+ }
+ assert(
+ !parquetBloomFilterPresent(nativeTable, "id") &&
+ !parquetBloomFilterPresent(jvmTable, "id"),
+ s"uninterpreted $suffix must not enable a Bloom filter")
+ }
+ }
+ }
+ }
+
+ test("max-bytes=32 falls back only when parquet-mr ignores the cap") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ val minimumBytes = 32
+ val naturallyMinimumNdv = 1
+ val bindingNdv = 1000000
+ val bindingFpp = 0.0001
+ val parquetMrBindingBytes = 4 * 1024 * 1024
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true', "
+ createTable(
+ "bloom_minimum_native",
+ partitionSpec = "",
+ properties = Some(
+ enabled + s"'write.parquet.bloom-filter-ndv.column.id'='$naturallyMinimumNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ createTable(
+ "bloom_minimum_fallback",
+ partitionSpec = "",
+ properties = Some(
+ enabled + s"'write.parquet.bloom-filter-fpp.column.id'='$bindingFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$bindingNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+
+ assertNativeWriteEngages("bloom_minimum_native", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_native VALUES (1, 'region', 1.0)")
+ }
+ assertNativeWriteDoesNotEngage("bloom_minimum_fallback", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_fallback VALUES (1, 'region', 1.0)")
+ }
+
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_native", "id").forall(_.length == minimumBytes))
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_fallback", "id")
+ .forall(_.length == parquetMrBindingBytes))
+ }
+ }
+
+ test("native bloom filters resolve list and map leaves to physical Parquet paths") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("native_nested_bloom", "native_nested_bloom_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY>,
+ attrs MAP>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.element.a'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.key'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.value.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10), named_struct('a', 20)),
+ map('small', named_struct('b', 30), 'large', named_struct('b', 40))),
+ (2, array(named_struct('a', 50)), map('medium', named_struct('b', 60)))
+ """)
+
+ assertNativeWriteEngages("native_nested_bloom", Seq(1, 2)) {
+ insert("native_nested_bloom")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("native_nested_bloom_jvm")
+ }
+
+ val physicalColumns =
+ Set("tags.list.element.a", "attrs.key_value.key", "attrs.key_value.value.b")
+ // Iceberg 1.5 passes the configured logical names directly to Parquet Java, where they do
+ // not match these physical list/map paths. Iceberg 1.8+ resolves canonical names first.
+ val expectedColumns =
+ if (icebergVersionAtLeast(1, 8)) physicalColumns else Set.empty[String]
+ assertParquetBloomFilters("native_nested_bloom", enabledColumns = expectedColumns)
+ assertParquetBloomFilters("native_nested_bloom_jvm", enabledColumns = expectedColumns)
+ }
+ }
+
+ test("short nested bloom aliases match the Iceberg runtime footer") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("nested_alias_native", "nested_alias_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY>,
+ attrs MAP>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.a'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10)), map('first', named_struct('b', 20))),
+ (2, array(named_struct('a', 30)), map('second', named_struct('b', 40)))
+ """)
+
+ assertNativeWriteEngages("nested_alias_native", Seq(1, 2)) {
+ insert("nested_alias_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("nested_alias_jvm")
+ }
+
+ val icebergUsesFieldIdResolution =
+ icebergVersionAtLeast(1, 8) && !icebergVersionAtLeast(1, 10)
+ Seq("tags.list.element.a", "attrs.key_value.value.b").foreach { physicalColumn =>
+ val nativeHasBloom = parquetBloomFilterPresent("nested_alias_native", physicalColumn)
+ val jvmHasBloom = parquetBloomFilterPresent("nested_alias_jvm", physicalColumn)
+ assert(nativeHasBloom == jvmHasBloom, s"native and JVM footer differ for $physicalColumn")
+ assert(
+ jvmHasBloom == icebergUsesFieldIdResolution,
+ s"unexpected short-alias behavior for $physicalColumn on this Iceberg runtime")
+ }
+ }
+ }
+
+ test("Iceberg 1.5 physical nested bloom paths use the JVM writer") {
+ assumeNativeAcceleration()
+ assume(
+ !icebergVersionAtLeast(1, 8),
+ "physical-path Bloom properties are specific to Iceberg 1.5")
+ withIcebergCatalog { _ =>
+ Seq("physical_path_native", "physical_path_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY>,
+ attrs MAP>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.list.element.a'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.key_value.value.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10)), map('first', named_struct('b', 20))),
+ (2, array(named_struct('a', 30)), map('second', named_struct('b', 40)))
+ """)
+
+ assertNativeWriteDoesNotEngage("physical_path_native", Seq(1, 2)) {
+ insert("physical_path_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("physical_path_jvm")
+ }
+
+ Seq("tags.list.element.a", "attrs.key_value.value.b").foreach { physicalColumn =>
+ assert(
+ parquetBloomFilterPresent("physical_path_native", physicalColumn),
+ s"expected preserved Bloom filter for $physicalColumn")
+ assert(
+ parquetBloomFilterPresent("physical_path_jvm", physicalColumn),
+ s"expected JVM Bloom filter for $physicalColumn")
+ }
+ }
+ }
+
+ test("unconfigured quoted columns preserve native Bloom writes") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.bloom_unconfigured_quoted (
+ id INT, `order id` INT
+ ) USING iceberg
+ TBLPROPERTIES ('write.parquet.bloom-filter-enabled.column.id'='true')
+ """)
+ assertNativeWriteEngages("bloom_unconfigured_quoted", 0 until 1000) {
+ spark.sql(s"""
+ INSERT INTO $catalog.$ns.bloom_unconfigured_quoted
+ SELECT CAST(id AS INT), CAST(id AS INT) FROM range(0, 1000, 1, 1)
+ """)
+ }
+ assert(parquetBloomFilterPresent("bloom_unconfigured_quoted", "id"))
+ assertParquetBloomContainsInts("bloom_unconfigured_quoted", "id", 0 until 1000)
+ assert(!parquetBloomFilterPresent("bloom_unconfigured_quoted", "order id"))
+ }
+ }
+
+ test("quoted bloom-filter columns renamed by Iceberg fall back to the JVM footer") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("bloom_quoted_name", "bloom_quoted_name_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ `order id` INT
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.order id'='true'
+ )
+ """)
+ }
+
+ val snapshot = withNativeEnabled {
+ captureWrite("bloom_quoted_name") {
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name VALUES (1), (2)")
+ }
+ }
+ assertExactlyOneCommit(snapshot)
+ val nativeExecs = snapshot.plans.flatMap { plan =>
+ collectWithSubqueries(plan) { case exec: CometIcebergWriteExec => exec }
+ }
+ assert(
+ nativeExecs.isEmpty,
+ s"expected the sanitized Bloom-filter path to fall back, plans:\n${snapshot.plans.mkString("\n--\n")}")
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name_jvm VALUES (1), (2)")
+ }
+ val fallbackHasBloom = parquetBloomFilterPresent("bloom_quoted_name", "order_x20id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_quoted_name_jvm", "order_x20id")
+ assert(
+ fallbackHasBloom == jvmHasBloom,
+ "expected the fallback write to match the Parquet Java footer")
+ if (icebergSupportsBloomFpp) {
+ assert(jvmHasBloom, "expected this Iceberg version to write the quoted-column filter")
+ }
+ }
+ }
+
+ test("native bloom filter is byte-identical to parquet-mr when max-bytes binds") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // This NDV/FPP pair requests a 128 MiB allocation before max-bytes is applied. The 4 KiB
+ // maximum must therefore bind on both writers rather than merely coinciding with the
+ // naturally selected size.
+ val configuredNdv = 100000000L
+ val bindingMaxBytes = 4 * 1024
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='0.01', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$bindingMaxBytes'")
+ createTable("bloom_identity_native", partitionSpec = "", properties)
+ createTable("bloom_identity_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) FROM range(0, 10000, 1, 1)")
+
+ assertNativeWriteEngages("bloom_identity_native", 0 until 10000) {
+ insert("bloom_identity_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_identity_jvm")
+ }
+
+ val native = parquetBloomFilterBytes("bloom_identity_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_identity_jvm", "id")
+ assert(
+ native.nonEmpty && native.forall(_.length == bindingMaxBytes),
+ s"native Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(
+ jvm.nonEmpty && jvm.forall(_.length == bindingMaxBytes),
+ s"JVM Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(native.size == jvm.size, "native and JVM writes must produce the same file count")
+ assert(
+ native.zip(jvm).forall { case (left, right) => java.util.Arrays.equals(left, right) },
+ "expected byte-identical capped SBBF bitsets for identical values and allocation")
+ }
+ }
+
+ test("native bloom filters match Java bytes and membership across physical types") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ case class BloomColumn(
+ name: String,
+ sqlType: String,
+ expression: String,
+ physicalType: String,
+ encodedValue: Int => Any)
+ val columns = Seq(
+ BloomColumn(
+ "big",
+ "BIGINT",
+ "(id - 5000) * 1000000007",
+ "INT64",
+ id => (id - 5000).toLong * 1000000007L),
+ BloomColumn(
+ "text",
+ "STRING",
+ "concat('bloom-', cast(id - 5000 AS STRING), 'é')",
+ "BINARY",
+ id => Binary.fromString(s"bloom-${id - 5000}é")),
+ BloomColumn(
+ "bytes",
+ "BINARY",
+ "unhex(concat('00FF', lpad(hex(id), 8, '0')))",
+ "BINARY",
+ id =>
+ Binary.fromConstantByteArray(
+ ByteBuffer
+ .allocate(6)
+ .put(0.toByte)
+ .put(0xff.toByte)
+ .putInt(id)
+ .array())),
+ BloomColumn(
+ "day",
+ "DATE",
+ "date_add(DATE '1970-01-01', cast(id - 5000 AS INT))",
+ "INT32",
+ id => id - 5000),
+ BloomColumn(
+ "ts",
+ "TIMESTAMP",
+ "timestamp_micros((id - 5000) * 1000003)",
+ "INT64",
+ id => (id - 5000).toLong * 1000003L),
+ BloomColumn(
+ "decimal_int32",
+ "DECIMAL(9, 2)",
+ "cast((id - 5000) / 100.0 AS DECIMAL(9, 2))",
+ "INT32",
+ id => id - 5000),
+ BloomColumn(
+ "decimal_int64",
+ "DECIMAL(18, 2)",
+ "cast((id - 5000) * 10000000000 AS DECIMAL(18, 2))",
+ "INT64",
+ id => (id - 5000).toLong * 1000000000000L),
+ BloomColumn(
+ "decimal_fixed",
+ "DECIMAL(38, 2)",
+ "cast((id - 5000) * 10000000000000000000000 AS DECIMAL(38, 2))",
+ "FIXED_LEN_BYTE_ARRAY",
+ id => {
+ val unscaled = BigInt(id - 5000) * BigInt(10).pow(24)
+ val bytes = unscaled.toByteArray
+ val padding = Array.fill[Byte](16 - bytes.length)(
+ if (unscaled.signum < 0) 0xff.toByte else 0.toByte)
+ Binary.fromConstantByteArray(padding ++ bytes)
+ }))
+ // Java can suppress filters for dictionary-only chunks. Force plain pages so every
+ // configured type contributes a filter to the cross-writer comparison.
+ val properties = (Seq(
+ "'write.parquet.bloom-filter-max-bytes'='4096'",
+ "'write.parquet.dict-size-bytes'='1'") ++ columns.map { column =>
+ s"'write.parquet.bloom-filter-enabled.column.${column.name}'='true'"
+ }).mkString(", ")
+ Seq("bloom_types_native", "bloom_types_jvm").foreach { table =>
+ val fields = columns.map(column => s"${column.name} ${column.sqlType}").mkString(", ")
+ spark.sql(
+ s"CREATE TABLE $catalog.$ns.$table (id INT, $fields) " +
+ s"USING iceberg TBLPROPERTIES ($properties)")
+ }
+ def insert(table: String): Unit = {
+ val values = columns.map(_.expression).mkString(", ")
+ spark.sql(
+ s"INSERT INTO $catalog.$ns.$table SELECT cast(id AS INT), $values " +
+ "FROM range(0, 10000, 1, 1)")
+ }
+ // No NDV/FPP properties: this fixture also runs on Iceberg 1.5.2. Ten thousand distinct
+ // values keep the 4 KiB filters dense enough to prevent Rust's post-insert folding.
+ // FLOAT/DOUBLE have separate fallback coverage because their Bloom writes use Java.
+ assertNativeWriteEngages("bloom_types_native", 0 until 10000)(insert("bloom_types_native"))
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_types_jvm")
+ }
+ columns.foreach { column =>
+ val native = parquetBloomFilterBytes("bloom_types_native", column.name)
+ val jvm = parquetBloomFilterBytes("bloom_types_jvm", column.name)
+ assert(native.size == 1 && jvm.size == 1, s"expected one filter for ${column.name}")
+ assert(native.head.length == 4096 && jvm.head.length == 4096)
+ assert(
+ java.util.Arrays.equals(native.head, jvm.head),
+ s"expected byte-identical Bloom filters for ${column.name}")
+ Seq("bloom_types_native", "bloom_types_jvm").foreach { table =>
+ assertParquetBloomContainsValues(
+ table,
+ column.name,
+ column.physicalType,
+ (0 until 10000).map(column.encodedValue))
+ }
+ }
+ }
+ }
+
+ test("native bloom sizing matches JVM for shape properties supported by the Iceberg runtime") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true'"
+ val cases: Seq[(String, String, Boolean, Boolean)] = Seq(
+ (
+ "fpp_only",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.02'",
+ icebergSupportsBloomFpp,
+ false),
+ (
+ "ndv_only",
+ s"$enabled, 'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomNdv,
+ true),
+ (
+ "both",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.005', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true),
+ // The requested NDV/FPP needs far more than 64 bytes. Like parquet-mr, max wins and the
+ // target FPP becomes impossible to guarantee, but membership must remain correct.
+ (
+ "binding_cap",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.0001', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000000', " +
+ "'write.parquet.bloom-filter-max-bytes'='64'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true))
+
+ cases.filter(_._3).foreach { case (suffix, properties, _, expectByteIdentity) =>
+ val nativeTable = s"bloom_shape_${suffix}_native"
+ val jvmTable = s"bloom_shape_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(table, partitionSpec = "", properties = Some(properties))
+ }
+ // Keep the explicit-NDV cases at their estimated cardinality so parquet-rs does not
+ // fold their allocation before byte-parity is checked.
+ val insertedCardinality = if (suffix == "ndv_only" || suffix == "both") 1000 else 256
+ val ids = 0 until insertedCardinality
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table SELECT CAST(id AS INT), 'region', " +
+ s"CAST(id AS DOUBLE) FROM range(${ids.start}, ${ids.end}, 1, 1)")
+
+ assertNativeWriteEngages(nativeTable, ids) {
+ insert(nativeTable)
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert(jvmTable)
+ }
+ val native = parquetBloomFilterBytes(nativeTable, "id")
+ val jvm = parquetBloomFilterBytes(jvmTable, "id")
+ if (expectByteIdentity) {
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ // Without explicit NDV, parquet-rs can fold to the observed cardinality while Parquet
+ // Java retains its initial maximum allocation. Folding is safe for readers, so require
+ // only that native uses no more space and still contains every inserted value.
+ assert(native.zip(jvm).forall { case (left, right) => left.length <= right.length })
+ assertParquetBloomContainsInts(nativeTable, "id", ids)
+ }
+ if (suffix == "binding_cap") {
+ assert(native.forall(_.length == 64), s"expected binding 64-byte cap for $nativeTable")
+ assertParquetBloomContainsInts(nativeTable, "id", ids)
+ }
+ }
+ }
+ }
+
+ test("explicit underestimated NDV retains parquet-mr allocation precedence") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // max-bytes is only a cap in parquet-mr; it does not enlarge a filter whose explicit NDV
+ // was underestimated. This comparison prevents Comet from silently replacing the user's
+ // NDV with an artificial NDV derived from the much larger maximum.
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='0.01', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='10', " +
+ "'write.parquet.bloom-filter-max-bytes'='67108864'")
+ createTable("bloom_low_ndv_native", partitionSpec = "", properties)
+ createTable("bloom_low_ndv_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) FROM range(0, 4096, 1, 1)")
+
+ assertNativeWriteEngages("bloom_low_ndv_native", 0 until 4096) {
+ insert("bloom_low_ndv_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_low_ndv_jvm")
+ }
+
+ val native = parquetBloomFilterBytes("bloom_low_ndv_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_low_ndv_jvm", "id")
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ }
+ }
+
+ test("overestimated explicit NDV folds and exposes the realized-FPP tradeoff") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ val configuredNdv = 1000L
+ val configuredFpp = 0.01
+ val insertedIds = 0 until 100
+ val absentProbeValues = 10000000 until 11000000
+ val expectedNativeBytes = 128
+ val expectedJvmBytes = 2048
+ val expectedNativeFalsePositives = 9088L
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ s"'write.parquet.bloom-filter-fpp.column.id'='$configuredFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'")
+ createTable("bloom_explicit_ndv_fold_native", partitionSpec = "", properties)
+ createTable("bloom_explicit_ndv_fold_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) " +
+ s"FROM range(${insertedIds.start}, ${insertedIds.end}, 1, 1)")
+
+ assertNativeWriteEngages("bloom_explicit_ndv_fold_native", insertedIds) {
+ insert("bloom_explicit_ndv_fold_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_explicit_ndv_fold_jvm")
+ }
+
+ val native = parquetBloomFilterBytes("bloom_explicit_ndv_fold_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_explicit_ndv_fold_jvm", "id")
+ assert(native.nonEmpty && native.size == jvm.size)
+ assert(
+ native.forall(_.length == expectedNativeBytes),
+ s"expected folded native filters of $expectedNativeBytes bytes")
+ assert(
+ jvm.forall(_.length == expectedJvmBytes),
+ s"expected Parquet Java filters of $expectedJvmBytes bytes")
+ assertParquetBloomContainsInts("bloom_explicit_ndv_fold_native", "id", insertedIds)
+ assertParquetBloomContainsInts("bloom_explicit_ndv_fold_jvm", "id", insertedIds)
+
+ val nativeFalsePositives =
+ parquetBloomFalsePositiveCount("bloom_explicit_ndv_fold_native", "id", absentProbeValues)
+ val jvmFalsePositives =
+ parquetBloomFalsePositiveCount("bloom_explicit_ndv_fold_jvm", "id", absentProbeValues)
+ assert(nativeFalsePositives == expectedNativeFalsePositives)
+ assert(jvmFalsePositives == 0L)
+ assert(nativeFalsePositives > jvmFalsePositives)
+ assert(nativeFalsePositives.toDouble / absentProbeValues.size <= configuredFpp)
+ }
+ }
+
+ test("large representable max folds natively while adjacent non-power-of-two falls back") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val powerOfTwo = 64 * 1024 * 1024
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true', "
+ createTable(
+ "bloom_large_native",
+ partitionSpec = "",
+ properties = Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='$powerOfTwo'"))
+ createTable(
+ "bloom_large_jvm",
+ partitionSpec = "",
+ properties =
+ Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='${powerOfTwo + 1}'"))
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) FROM range(0, 256, 1, 1)")
+
+ assertNativeWriteEngages("bloom_large_native", 0 until 256) {
+ insert("bloom_large_native")
+ }
+ assertNativeWriteDoesNotEngage("bloom_large_jvm", 0 until 256) {
+ insert("bloom_large_jvm")
+ }
+
+ val nativeBytes = parquetBloomFilterBytes("bloom_large_native", "id")
+ val nativeSize = nativeBytes.head.length
+ val jvmSize = parquetBloomFilterBytes("bloom_large_jvm", "id").head.length
+ assert(nativeSize < 1024 * 1024, s"expected folding, got $nativeSize bytes")
+ // Iceberg runtimes bundle different Parquet Java versions: newer versions retain the exact
+ // cap while older ones round it upward. Both demonstrate that the fallback avoids folding
+ // this deliberately oversized filter down to the native allocation.
+ assert(jvmSize > powerOfTwo, s"expected an oversized JVM filter, got $jvmSize bytes")
+ assertParquetBloomContainsInts("bloom_large_native", "id", 0 until 256)
+
+ // The adjacent non-power-of-two JVM filter is intentionally much larger, so comparing its
+ // bytes with the folded native filter would be meaningless. Instead, write the same values
+ // through the JVM writer with a cap equal to the observed folded size. Power-of-two folding
+ // must produce exactly the same SBBF bitset as hashing directly into that final allocation.
+ createTable(
+ "bloom_large_folded_jvm",
+ partitionSpec = "",
+ properties = Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='$nativeSize'"))
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_large_folded_jvm")
+ }
+ assertRows("bloom_large_folded_jvm", 0 until 256)
+ val foldedJvmBytes = parquetBloomFilterBytes("bloom_large_folded_jvm", "id")
+ assert(nativeBytes.map(_.length) == foldedJvmBytes.map(_.length))
+ assert(
+ nativeBytes.zip(foldedJvmBytes).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ },
+ "folded native SBBF must be byte-identical to a same-sized JVM SBBF")
+ }
+ }
+
+ test("out-of-range bloom max uses the classic writer when a bloom column is configured") {
+ assumeNativeBloomRuntime()
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // A tiny explicit NDV keeps the fallback writer's allocation small even for the oversized
+ // maximum. Without a configured Bloom column, the maximum is unused and stays native.
+ val enabledWithSmallNdv =
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1', "
+ val max = "134217729"
+ val table = "bloom_range_fallback"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(enabledWithSmallNdv + s"'write.parquet.bloom-filter-max-bytes'='$max'"))
+ assertNativeWriteDoesNotEngage(table, Seq(1)) {
+ spark.sql(s"INSERT INTO cat.db.$table VALUES (1, 'region', 1.0)")
+ }
+ }
+ }
+
// What Iceberg's Spark writer stamps for `sort_order_id` on appended files changed across
// releases: through 1.10 `SparkWrite$WriterFactory` never wires the table sort order (files
// get 0 even on a sorted table); 1.11 added `SparkWriteConf.outputSortOrderId` and stamps the
@@ -4821,11 +5676,217 @@ class CometIcebergWriteActionSuite
assert(ids == expectedIds, s"expected $expectedIds, got $ids")
}
+ /**
+ * Reads every current data-file footer and verifies that parquet-rs emitted bloom-filter data
+ * exactly for the configured columns. Checking the plan alone would not catch a translation bug
+ * that selected `CometIcebergWriteExec` but silently omitted the bloom writer properties.
+ */
+ private def assertParquetBloomFilters(tableName: String, enabledColumns: Set[String]): Unit = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files")
+ .collect()
+ .map(_.getString(0))
+ .toSeq
+ assert(paths.nonEmpty, s"expected $tableName to have at least one data file")
+
+ val conf = spark.sparkContext.hadoopConfiguration
+ paths.foreach { path =>
+ val input = HadoopInputFile.fromPath(new HadoopPath(path.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ val columns = reader.getFooter.getBlocks.asScala.flatMap(_.getColumns.asScala)
+ assert(columns.nonEmpty, s"expected at least one column chunk in $path")
+ columns.foreach { column =>
+ val columnName = column.getPath.toDotString
+ val bloomFilter = reader.readBloomFilter(column)
+ if (enabledColumns.contains(columnName)) {
+ assert(
+ bloomFilter != null && bloomFilter.getBitsetSize > 0,
+ s"expected a non-empty bloom filter for $columnName in $path")
+ } else {
+ assert(
+ bloomFilter == null,
+ s"expected no bloom filter for unconfigured column $columnName in $path")
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ }
+
+ private def parquetBloomFilterBytes(tableName: String, columnName: String): Seq[Array[Byte]] = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files ORDER BY file_path")
+ .collect()
+ .map(_.getString(0))
+ val conf = spark.sparkContext.hadoopConfiguration
+ paths.toSeq.flatMap { path =>
+ val input = HadoopInputFile.fromPath(new HadoopPath(path.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ reader.getFooter.getBlocks.asScala.flatMap { block =>
+ block.getColumns.asScala
+ .filter(_.getPath.toDotString == columnName)
+ .map { column =>
+ val bloom = reader.readBloomFilter(column)
+ assert(bloom != null, s"expected Bloom filter for $columnName in $path")
+ val out = new ByteArrayOutputStream(bloom.getBitsetSize)
+ bloom.writeTo(out)
+ out.toByteArray
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ }
+
+ private def parquetBloomFilterPresent(tableName: String, columnName: String): Boolean = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files ORDER BY file_path")
+ .collect()
+ .map(_.getString(0))
+ val conf = spark.sparkContext.hadoopConfiguration
+ val present = paths.toSeq.flatMap { path =>
+ val input = HadoopInputFile.fromPath(new HadoopPath(path.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ reader.getFooter.getBlocks.asScala.flatMap { block =>
+ block.getColumns.asScala
+ .filter(_.getPath.toDotString == columnName)
+ .map(column => reader.readBloomFilter(column) != null)
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ assert(present.nonEmpty, s"expected Parquet column $columnName in $tableName")
+ present.forall(identity)
+ }
+
+ private def parquetBloomFalsePositiveCount(
+ tableName: String,
+ columnName: String,
+ absentValues: Seq[Int]): Long = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files ORDER BY file_path")
+ .collect()
+ .map(_.getString(0))
+ val conf = spark.sparkContext.hadoopConfiguration
+ val counts = paths.toSeq.flatMap { path =>
+ val input = HadoopInputFile.fromPath(new HadoopPath(path.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ reader.getFooter.getBlocks.asScala.flatMap { block =>
+ block.getColumns.asScala
+ .filter(_.getPath.toDotString == columnName)
+ .map { column =>
+ val bloom = reader.readBloomFilter(column)
+ assert(bloom != null, s"expected Bloom filter for $columnName in $path")
+ absentValues.count(value => bloom.findHash(bloom.hash(value))).toLong
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ assert(counts.size == 1, s"expected one Bloom filter for $columnName, got ${counts.size}")
+ counts.head
+ }
+
+ private def assertParquetBloomContainsValues(
+ tableName: String,
+ columnName: String,
+ physicalType: String,
+ values: Seq[Any]): Unit = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files")
+ .collect()
+ .map(_.getString(0))
+ assert(paths.length == 1, s"expected one data file for $tableName")
+ val conf = spark.sparkContext.hadoopConfiguration
+ val input = HadoopInputFile.fromPath(new HadoopPath(paths.head.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ val chunks = reader.getFooter.getBlocks.asScala
+ .flatMap(_.getColumns.asScala)
+ .filter(_.getPath.toDotString == columnName)
+ assert(chunks.size == 1, s"expected one column chunk for $columnName in $tableName")
+ val column = chunks.head
+ assert(column.getPrimitiveType.getPrimitiveTypeName.name() == physicalType)
+ val bloom = reader.readBloomFilter(column)
+ assert(bloom != null, s"expected Bloom filter for $columnName in $tableName")
+ values.foreach { value =>
+ val hash = value match {
+ case v: Int => bloom.hash(v)
+ case v: Long => bloom.hash(v)
+ case v: Binary => bloom.hash(v)
+ case other => fail(s"unsupported Bloom membership value: $other")
+ }
+ assert(bloom.findHash(hash), s"Bloom false negative for $columnName=$value in $tableName")
+ }
+ } finally {
+ reader.close()
+ }
+ }
+
+ private def assertParquetBloomContainsInts(
+ tableName: String,
+ columnName: String,
+ values: Seq[Int]): Unit = {
+ val paths = spark
+ .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files")
+ .collect()
+ .map(_.getString(0))
+ val conf = spark.sparkContext.hadoopConfiguration
+ paths.foreach { path =>
+ val input = HadoopInputFile.fromPath(new HadoopPath(path.stripPrefix("file:")), conf)
+ val reader = ParquetFileReader.open(input)
+ try {
+ reader.getFooter.getBlocks.asScala.foreach { block =>
+ block.getColumns.asScala
+ .filter(_.getPath.toDotString == columnName)
+ .foreach { column =>
+ val bloom = reader.readBloomFilter(column)
+ assert(bloom != null, s"expected Bloom filter for $columnName in $path")
+ values.foreach { value =>
+ assert(
+ bloom.findHash(bloom.hash(value)),
+ s"Bloom filter false negative for $columnName=$value in $path")
+ }
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ }
+
/** Native acceleration shared assumption -- currently just the Iceberg-on-classpath check. */
private def assumeNativeAcceleration(): Unit = {
assume(icebergAvailable, "Iceberg not available in classpath")
}
+ private def assumeNativeBloomRuntime(): Unit =
+ assume(icebergVersionAtLeast(1, 10), "Native Bloom writes require Iceberg 1.10 or later")
+
+ private def assumeIcebergBloomShapeProperties(): Unit = {
+ assume(
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ "Iceberg runtime does not interpret per-column Bloom FPP/NDV properties")
+ }
+
+ private def icebergSupportsBloomFpp: Boolean =
+ org.apache.comet.iceberg.IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX")
+ .isDefined
+
+ private def icebergSupportsBloomNdv: Boolean =
+ org.apache.comet.iceberg.IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX")
+ .isDefined
+
/**
* Every row-consuming operator must receive row-based input. Spark guarantees that by inserting
* `ColumnarToRow` transitions in `ApplyColumnarRulesAndInsertTransitions`; an operator that
@@ -4862,10 +5923,12 @@ class CometIcebergWriteActionSuite
* version / session-state combinations where `withSQLConf` loses the override before the rule
* fires.
*/
- private def withNativeEnabled[T](action: => T): T =
+ private def withNativeEnabled[T](action: => T, bloomFilterFoldingEnabled: Boolean = true): T =
withSessionConf(
CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> Some("true"),
- CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> Some("true"))(action)
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> Some("true"),
+ CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key ->
+ Some(bloomFilterFoldingEnabled.toString))(action)
/**
* Sets each key to its value, or unsets it for `None`, for the duration of `action`, then puts
@@ -5093,7 +6156,15 @@ class CometIcebergWriteActionSuite
*/
private def assertNativeWriteDoesNotEngage(tableName: String, expectedIds: Seq[Int])(
action: => Unit): Unit = {
- val snapshot = withNativeEnabled { captureWrite(tableName)(action) }
+ assertNativeWriteDoesNotEngage(tableName, expectedIds, bloomFilterFoldingEnabled = true)(
+ action)
+ }
+
+ private def assertNativeWriteDoesNotEngage(
+ tableName: String,
+ expectedIds: Seq[Int],
+ bloomFilterFoldingEnabled: Boolean)(action: => Unit): Unit = {
+ val snapshot = withNativeEnabled(captureWrite(tableName)(action), bloomFilterFoldingEnabled)
assertExactlyOneCommit(snapshot)
val nativeExecs = snapshot.plans.flatMap { p =>
collectWithSubqueries(p) { case e: CometIcebergWriteExec => e }
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index a102d142c83..681b6da0de9 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -54,6 +54,7 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
override protected def sparkConf: SparkConf = {
super.sparkConf
.set(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key, "true")
+ .set(CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key, "true")
}
override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
@@ -377,26 +378,312 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- test("fall-back: write.parquet.bloom-filter-max-bytes set") {
+ test("legacy runtime rejects every configured Bloom entry before filtering disabled names") {
+ assume(!icebergVersionAtLeast(1, 10), "legacy runtime only")
withDetectionCatalog { _ =>
+ Seq(
+ ("id", "true"),
+ ("id", "false"),
+ ("", "true"),
+ ("", "false"),
+ ("missing", "true")).zipWithIndex.foreach { case ((column, enabled), index) =>
+ val table = s"legacy_bloom_setting_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-enabled.column.$column'='$enabled'"))
+ assertUnsupportedContainsAllowingWriteFailure(table, "Iceberg 1.10 or later")
+ }
+ }
+ }
+
+ test("legacy runtime also falls back for standalone Bloom shape properties") {
+ assume(!icebergVersionAtLeast(1, 10), "legacy runtime only")
+ withDetectionCatalog { _ =>
+ Seq("fpp", "ndv").zipWithIndex.foreach { case (shape, index) =>
+ val table = s"legacy_bloom_shape_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-$shape.column.id'='garbage'"))
+ assertUnsupportedContainsAllowingWriteFailure(table, "Iceberg 1.10 or later")
+ }
+ }
+ }
+
+ test("bloom-filter max-bytes accepts only representable power-of-two values") {
+ withDetectionCatalog { dir =>
+ Seq("32", "524288", "1048576", "134217728").zipWithIndex.foreach { case (value, index) =>
+ val table = s"bloom_max_ok_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$value'"))
+ assertSupportLevelIs[Compatible](table)
+ }
+
+ Seq("31", "33", "100", "134217729", "0", "-1", " 32", "garbage", "2147483648").zipWithIndex
+ .foreach { case (value, index) =>
+ val table = s"bloom_max_bad_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$value'"))
+ assertUnsupportedContainsAllowingWriteFailure(
+ table,
+ if (icebergVersionAtLeast(1, 10)) "write.parquet.bloom-filter-max-bytes"
+ else "Iceberg 1.10 or later")
+ }
+ }
+ }
+
+ test("unused bloom-filter max-bytes does not force fallback") {
+ withDetectionCatalog { dir =>
+ val nonPowerOfTwoMaxBytes = 100
createTable(
- "bloom_max",
+ "unused_bloom_max",
partitionSpec = "",
- properties = Some("'write.parquet.bloom-filter-max-bytes'='524288'"))
- assertUnsupportedContains("bloom_max", "write.parquet.bloom-filter-max-bytes")
+ properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$nonPowerOfTwoMaxBytes'"))
+ assertSupportLevelIs[Compatible]("unused_bloom_max")
}
}
- test("fall-back: per-column bloom filter enabled") {
+ test("unused bloom-filter max-bytes preserves Java validation") {
+ withDetectionCatalog { _ =>
+ Seq("0", "-1").zipWithIndex.foreach { case (value, index) =>
+ val table = s"unused_bloom_invalid_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$value'"))
+ assertUnsupportedContainsAllowingWriteFailure(
+ table,
+ "write.parquet.bloom-filter-max-bytes")
+ }
+ }
+ }
+
+ test("bloom-filter max-bytes=32 falls back only when parquet-mr ignores the cap") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ val minimumBytes = 32
+ val naturallyMinimumNdv = 1
+ val bindingFpp = 0.0001
+ val bindingNdv = 1000000
+ createTable(
+ "without_ndv",
+ partitionSpec = "",
+ properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ assertSupportLevelIs[Compatible]("without_ndv")
+
+ createTable(
+ "natural_minimum",
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$naturallyMinimumNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ assertSupportLevelIs[Compatible]("natural_minimum")
+
+ createTable(
+ "ignored_minimum_cap",
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ s"'write.parquet.bloom-filter-fpp.column.id'='$bindingFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$bindingNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ if (icebergSupportsBloomFpp && icebergSupportsBloomNdv) {
+ assertUnsupportedContainsAllowingWriteFailure(
+ "ignored_minimum_cap",
+ "write.parquet.bloom-filter-max-bytes")
+ } else {
+ assertSupportLevelIs[Compatible]("ignored_minimum_cap")
+ }
+ }
+ }
+
+ test("fall-back: bloom-filter NDV values that overflow parquet-mr sizing arithmetic") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ val largestNonOverflowingNdv = Long.MaxValue / 8L
+ val cases = Seq(
+ (largestNonOverflowingNdv, true),
+ (largestNonOverflowingNdv + 1L, false),
+ (1L << 61, false),
+ (Long.MaxValue, false))
+
+ cases.zipWithIndex.foreach { case ((ndv, compatible), index) =>
+ val table = s"bloom_ndv_overflow_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$ndv'"))
+ if (compatible || !icebergSupportsBloomNdv) {
+ assertSupportLevelIs[Compatible](table)
+ } else {
+ assertUnsupportedContainsAllowingWriteFailure(
+ table,
+ "write.parquet.bloom-filter-ndv.column.id")
+ }
+ }
+ }
+ }
+
+ test("bloom-filter enabled=false with NDV follows runtime support and validates NDV") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ val configuredNdv = 1000
+ createTable(
+ "false_with_ndv",
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'"))
+ assertSupportLevelIs[Compatible]("false_with_ndv")
+
+ createTable(
+ "false_with_bad_ndv",
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='garbage'"))
+ if (icebergSupportsBloomNdv) {
+ assertUnsupportedContainsAllowingWriteFailure(
+ "false_with_bad_ndv",
+ "write.parquet.bloom-filter-ndv")
+ } else {
+ assertSupportLevelIs[Compatible]("false_with_bad_ndv")
+ }
+ }
+ }
+
+ test("bloom-filter enabled=false still validates FPP") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ createTable(
+ "false_with_bad_fpp",
+ partitionSpec = "",
+ properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='garbage'"))
+ if (icebergSupportsBloomFpp) {
+ assertUnsupportedContainsAllowingWriteFailure(
+ "false_with_bad_fpp",
+ "write.parquet.bloom-filter-fpp")
+ } else {
+ assertSupportLevelIs[Compatible]("false_with_bad_fpp")
+ }
+ }
+ }
+
+ test("fall-back: invalid enabled-column FPP and NDV") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ Seq(
+ ("'write.parquet.bloom-filter-fpp.column.id'='0'", icebergSupportsBloomFpp),
+ ("'write.parquet.bloom-filter-fpp.column.id'='1'", icebergSupportsBloomFpp),
+ ("'write.parquet.bloom-filter-fpp.column.id'='NaN'", icebergSupportsBloomFpp),
+ // Positive and below one, but too small for any integer NDV to encode the requested
+ // power-of-two allocation through parquet-rs's NDV/FPP API.
+ ("'write.parquet.bloom-filter-fpp.column.id'='4.9E-324'", icebergSupportsBloomFpp),
+ ("'write.parquet.bloom-filter-ndv.column.id'='0'", icebergSupportsBloomNdv),
+ (
+ "'write.parquet.bloom-filter-ndv.column.id'='garbage'",
+ icebergSupportsBloomNdv)).zipWithIndex
+ .foreach { case ((property, interpreted), index) =>
+ val table = s"bloom_shape_bad_$index"
+ createTable(
+ table,
+ partitionSpec = "",
+ properties =
+ Some(s"'write.parquet.bloom-filter-enabled.column.id'='true', $property"))
+ if (interpreted) {
+ assertUnsupportedContainsAllowingWriteFailure(table, "write.parquet.bloom-filter")
+ } else {
+ assertSupportLevelIs[Compatible](table)
+ }
+ }
+ }
+ }
+
+ test("Compatible when a per-column bloom filter is enabled") {
+ assumeNativeBloomRuntime()
withDetectionCatalog { _ =>
createTable(
"bloom_col",
partitionSpec = "",
properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
- assertUnsupportedContains(
- "bloom_col",
- "write.parquet.bloom-filter-enabled.column.id",
- "true")
+ assertSupportLevelIs[Compatible]("bloom_col")
+ }
+ }
+
+ test("floating-point Bloom detection reports top-level and nested affected leaves") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { _ =>
+ val columns = Seq(
+ "amount" -> "amount",
+ "f" -> "f",
+ "detail.value" -> "detail.value",
+ "samples.element" -> "samples.list.element",
+ "attrs.value" -> "attrs.key_value.value",
+ "lookup.key" -> "lookup.key_value.key")
+ columns.zipWithIndex.foreach { case ((logical, physical), index) =>
+ val column = if (icebergSupportsBloomFpp) logical else physical
+ val table = s"floating_bloom_$index"
+ val write = floatingBloomWriteExec(
+ table,
+ s"'write.parquet.bloom-filter-enabled.column.$column'='true'")
+ assertUnsupportedContains(write, table, "FLOAT/DOUBLE", column, "NaN equality")
+ }
+ }
+ }
+
+ test("floating-point Bloom detection preserves native eligibility for unaffected columns") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { _ =>
+ val stringKey = if (icebergSupportsBloomFpp) "attrs.key" else "attrs.key_value.key"
+ val integerValue = if (icebergSupportsBloomFpp) "lookup.value" else "lookup.key_value.value"
+ val cases = Seq(
+ "'write.parquet.bloom-filter-enabled.column.id'='true'",
+ s"'write.parquet.bloom-filter-enabled.column.$stringKey'='true'",
+ s"'write.parquet.bloom-filter-enabled.column.$integerValue'='true'",
+ "'write.parquet.bloom-filter-enabled.column.missing'='true'",
+ "'write.parquet.bloom-filter-enabled.column.amount'='false'")
+ cases.zipWithIndex.foreach { case (properties, index) =>
+ val write = floatingBloomWriteExec(s"unaffected_bloom_$index", properties)
+ assert(CometIcebergNativeWrite.getSupportLevel(write).isInstanceOf[Compatible])
+ }
+ val table = "floating_bloom_false_ndv"
+ val write = floatingBloomWriteExec(
+ table,
+ "'write.parquet.bloom-filter-enabled.column.amount'='false', " +
+ "'write.parquet.bloom-filter-ndv.column.amount'='1000'")
+ if (icebergSupportsBloomNdv) {
+ assertUnsupportedContains(write, table, "FLOAT/DOUBLE", "amount", "NaN equality")
+ } else {
+ assert(CometIcebergNativeWrite.getSupportLevel(write).isInstanceOf[Compatible])
+ }
+ }
+ }
+
+ test("Bloom-filter folding must be explicitly enabled for native writes") {
+ assumeNativeBloomRuntime()
+ withDetectionCatalog { dir =>
+ createTable(
+ "bloom_folding_opt_in",
+ partitionSpec = "",
+ properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
+ withSQLConf(
+ CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key -> "false") {
+ assertUnsupportedContains(
+ "bloom_folding_opt_in",
+ CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key)
+ }
}
}
@@ -1595,6 +1882,20 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
+ private def floatingBloomWriteExec(tableName: String, properties: String): IcebergWriteExec = {
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$tableName (
+ id INT, region STRING, amount DOUBLE, f FLOAT,
+ detail STRUCT, samples ARRAY,
+ attrs MAP, lookup MAP
+ ) USING iceberg TBLPROPERTIES ($properties)
+ """)
+ planInsertWriteExec(
+ s"$catalog.$ns.$tableName",
+ "(1, 'region', 1.0, CAST(2.0 AS FLOAT), named_struct('value', 3.0D), " +
+ "array(CAST(4.0 AS FLOAT)), map('key', 5.0D), map(CAST(6.0 AS FLOAT), 1))")
+ }
+
private def createTable(
tableName: String,
partitionSpec: String,
@@ -1778,6 +2079,19 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
+ private def assumeNativeBloomRuntime(): Unit =
+ assume(icebergVersionAtLeast(1, 10), "Native Bloom writes require Iceberg 1.10 or later")
+
+ private def icebergSupportsBloomFpp: Boolean =
+ IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX")
+ .isDefined
+
+ private def icebergSupportsBloomNdv: Boolean =
+ IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX")
+ .isDefined
+
/**
* Runs Spark's transition insertion followed by [[EliminateRedundantTransitions]] over a
* hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source` plan and returns the
diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala
index f1dd09ccf7e..cecf245e8c4 100644
--- a/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala
@@ -19,8 +19,11 @@
package org.apache.comet.serde.operator
+import scala.jdk.CollectionConverters._
+
import org.scalatest.funsuite.AnyFunSuite
+import org.apache.comet.iceberg.IcebergReflection
import org.apache.comet.serde.OperatorOuterClass.{CompressionCodec, IcebergWriterMode}
/**
@@ -33,6 +36,48 @@ class IcebergWriteProtoTranslationSuite extends AnyFunSuite {
private val TestCreatedBy = "Apache DataFusion Comet (test)"
+ test("Bloom path resolution rejects unidentified Iceberg naming conventions") {
+ val resolution = IcebergReflection.ParquetPathResolution(
+ Map("tags.element.a" -> "tags.list.element.a"),
+ Map(1 -> "tags.list.element.a"),
+ Set("tags.list.element.a"),
+ Set.empty,
+ Set.empty)
+ Seq("unknown", "", "vendor-build", "1.8.bad", "1.10", "2.0.0").foreach { version =>
+ val error = intercept[IllegalArgumentException] {
+ CometIcebergNativeWrite.bloomFilterParquetPaths(Seq("tags.a"), resolution, version)
+ }
+ assert(error.getMessage.contains("Cannot determine Iceberg Bloom column-name convention"))
+ assert(error.getMessage.contains(s"'$version'"))
+ }
+ }
+
+ test("Bloom path resolution preserves recognized runtime naming conventions") {
+ val physicalPath = "tags.list.element.a"
+ val resolution = IcebergReflection.ParquetPathResolution(
+ Map("tags.element.a" -> physicalPath, "order id" -> "order_x20id"),
+ Map(1 -> physicalPath),
+ Set(physicalPath, "order_x20id"),
+ Set("order_x20id"),
+ Set.empty)
+ Seq("1.10.0", "1.11.0", "1.11.0-SNAPSHOT", "1.11.0+vendor").foreach { version =>
+ assert(
+ CometIcebergNativeWrite.bloomFilterParquetPaths(
+ Seq("tags.element.a", "tags.a"),
+ resolution,
+ version) == Map("tags.element.a" -> physicalPath))
+ }
+ }
+
+ test("Bloom writes reject older runtimes even when the version is identifiable") {
+ Seq("1.5.2", "1.8.1", "1.9.1", "1.8.1-SNAPSHOT").foreach { version =>
+ assert(
+ CometIcebergNativeWrite
+ .bloomFilterRuntimeRejection(version)
+ .exists(_.contains("Iceberg 1.10 or later")))
+ }
+ }
+
// -- compression -----------------------------------------------------------
test("compression defaults to zstd at parquet-mr's level 3 when no property is set") {
@@ -176,6 +221,112 @@ class IcebergWriteProtoTranslationSuite extends AnyFunSuite {
assert(settings.getPageRowLimit == 1000)
}
+ test("per-column bloom filter properties are translated deterministically") {
+ val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix
+ val settings = buildParquetSettings(
+ Map(
+ s"${prefix}region" -> "TRUE",
+ s"${prefix}id" -> "true",
+ s"${prefix}amount" -> "false",
+ s"${Keys.ParquetBloomFilterColumnFppPrefix}region" -> "0.02",
+ s"${Keys.ParquetBloomFilterColumnNdvPrefix}id" -> "1234"),
+ TestCreatedBy)
+ assert(settings.getBloomFiltersList.asScala.map(_.getParquetPath) == Seq("id", "region"))
+ val bloomFilters = settings.getBloomFiltersList.asScala.map(p => p.getParquetPath -> p).toMap
+ assert(bloomFilters.keySet == Set("id", "region"))
+ assert(settings.getBloomFilterMaxBytes == 1024L * 1024L)
+ assert(bloomFilters("id").getFpp == 0.01d)
+ assert(bloomFilters("region").getFpp == 0.02d)
+ assert(bloomFilters("id").getNdv == 1234L)
+ assert(!bloomFilters("region").hasNdv)
+ assert(!bloomFilters.contains("amount"))
+ }
+
+ test("an explicit NDV re-enables a bloom filter after enabled=false") {
+ val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix
+ val configuredFpp = 0.02d
+ val configuredNdv = 1234L
+ val settings = buildParquetSettings(
+ Map(
+ s"${prefix}id" -> "false",
+ s"${Keys.ParquetBloomFilterColumnFppPrefix}id" -> configuredFpp.toString,
+ s"${Keys.ParquetBloomFilterColumnNdvPrefix}id" -> configuredNdv.toString),
+ TestCreatedBy)
+
+ val bloomFilter = settings.getBloomFilters(0)
+ assert(bloomFilter.getParquetPath == "id")
+ assert(bloomFilter.getFpp == configuredFpp)
+ assert(bloomFilter.getNdv == configuredNdv)
+ }
+
+ test("bloom filter properties use physical Parquet paths for list and map leaves") {
+ val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix
+ val settings = buildParquetSettings(
+ Map(
+ s"${prefix}tags.element" -> "true",
+ s"${prefix}attrs.key" -> "true",
+ s"${prefix}attrs.value" -> "true",
+ s"${prefix}missing" -> "true",
+ s"${Keys.ParquetBloomFilterColumnFppPrefix}tags.element" -> "0.02",
+ s"${Keys.ParquetBloomFilterColumnNdvPrefix}attrs.value" -> "1234"),
+ TestCreatedBy,
+ Map(
+ "tags.element" -> "tags.list.element",
+ "attrs.key" -> "attrs.key_value.key",
+ "attrs.value" -> "attrs.key_value.value"))
+
+ val bloomFilters = settings.getBloomFiltersList.asScala.map(p => p.getParquetPath -> p).toMap
+ assert(
+ bloomFilters.keySet ==
+ Set("attrs.key_value.key", "attrs.key_value.value", "tags.list.element"))
+ assert(bloomFilters("tags.list.element").getFpp == 0.02d)
+ assert(bloomFilters("attrs.key_value.value").getNdv == 1234L)
+ assert(!bloomFilters.contains("missing"))
+ }
+
+ test("Iceberg bloom filter defaults and explicit max are translated exactly") {
+ val enabled = Keys.ParquetBloomFilterColumnEnabledPrefix + "id"
+ val defaults = buildParquetSettings(Map(enabled -> "true"), TestCreatedBy)
+ assert(defaults.getBloomFilterMaxBytes == Defaults.BloomFilterMaxBytes)
+ assert(Defaults.BloomFilterMaxBytes == 1024 * 1024)
+ assert(defaults.getBloomFilters(0).getFpp == Defaults.BloomFilterFpp)
+ assert(Defaults.BloomFilterFpp == 0.01d)
+
+ val explicit = buildParquetSettings(
+ Map(enabled -> "true", Keys.ParquetBloomFilterMaxBytes -> "67108864"),
+ TestCreatedBy)
+ assert(explicit.getBloomFilterMaxBytes == 64L * 1024L * 1024L)
+ }
+
+ test("Bloom sizing arithmetic stays aligned with the native writer") {
+ import CometIcebergNativeWrite.{bloomFilterSizesRepresentable, MinBloomFilterBytes, parquetMrBloomFilterBytes}
+
+ val kibibyte = 1024
+ val mebibyte = kibibyte * kibibyte
+ val twiceMinimum = 2 * MinBloomFilterBytes
+ // Keep this table in sync with BLOOM_FILTER_SIZING_CASES in iceberg_write.rs. These fixed
+ // expectations make JVM planning and native task sizing fail independently if either drifts.
+ val cases = Seq(
+ (None, 0.01d, mebibyte, mebibyte),
+ (Some(1L), 0.01d, mebibyte, MinBloomFilterBytes),
+ (Some(1000L), 0.01d, mebibyte, 2 * kibibyte),
+ (Some(1000L), 0.005d, mebibyte, 2 * kibibyte),
+ (Some(1000000L), 0.0001d, twiceMinimum, twiceMinimum),
+ (Some(100000000L), 0.01d, 4 * kibibyte, 4 * kibibyte))
+
+ cases.foreach { case (ndv, fpp, maxBytes, expectedBytes) =>
+ assert(
+ parquetMrBloomFilterBytes(ndv, fpp, maxBytes) == expectedBytes,
+ s"ndv=$ndv fpp=$fpp maxBytes=$maxBytes")
+ assert(
+ bloomFilterSizesRepresentable(maxBytes, fpp),
+ s"ndv=$ndv fpp=$fpp maxBytes=$maxBytes")
+ }
+ assert(
+ !bloomFilterSizesRepresentable(mebibyte, java.lang.Double.MIN_NORMAL),
+ "the gate must reject a target for which native synthetic-NDV sizing fails")
+ }
+
test("size properties are parsed with Java Integer.parseInt semantics") {
// No trimming and no values past Int.MaxValue -- exactly what iceberg-java's
// PropertyUtil.propertyAsInt would do. The eligibility gate declines these values
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala
index 3ca224ec1a4..750cb81b018 100644
--- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala
@@ -20,7 +20,11 @@
package org.apache.spark.sql.benchmark
import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+import org.apache.hadoop.fs.{Path => HadoopPath}
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.util.HadoopInputFile
import org.apache.spark.SparkConf
import org.apache.spark.benchmark.Benchmark
import org.apache.spark.sql.SparkSession
@@ -61,8 +65,8 @@ import org.apache.comet.CometSparkSessionExtensions
* pay equally. It dilutes the writer's factor rather than inflating it.
* - The warehouse is a local temporary directory, so no object-store latency is included.
* - The two writers choose different file roll points (see the accepted divergences in
- * `docs/source/user-guide/latest/iceberg-writes.md`), so the resulting file layouts are not
- * expected to match. Only wall clock and row counts are compared.
+ * `docs/source/user-guide/latest/iceberg.md`), so the resulting file layouts are not expected
+ * to match. Only wall clock and row counts are compared.
* - `Rate` and `Per Row` are always computed against the corpus size, so for the `DELETE` case
* they describe the rows the statement passed over, not the far smaller number it removed.
* - The corpus below is this benchmark's own, not the one #5361 measured. The factors are the
@@ -72,7 +76,10 @@ import org.apache.comet.CometSparkSessionExtensions
* {{{
* SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometIcebergWriteBenchmark
* }}}
- * Results will be written to "spark/benchmarks/CometIcebergWriteBenchmark-**results.txt".
+ * Results will be written to "spark/benchmarks/CometIcebergWriteBenchmark-**results.txt". Pass
+ * `bloom`, `bloom-off`, or `bloom-on`, followed optionally by row count and iteration count, to
+ * compare native writes without and with Bloom filters. The contributor benchmarking guide
+ * describes collecting peak RSS in separate JVMs with the operating system's time utility.
*/
object CometIcebergWriteBenchmark extends CometBenchmarkBase {
@@ -285,6 +292,16 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
}
override def runCometBenchmark(mainArgs: Array[String]): Unit = {
+ val mode = mainArgs.headOption.getOrElse("all")
+ require(
+ Set("all", "bloom", "bloom-off", "bloom-on").contains(mode),
+ "Expected all, bloom, bloom-off, or bloom-on, optionally followed by rows and iterations")
+ require(mainArgs.length <= 3, "Expected at most mode, rows, and iterations")
+ val rows = mainArgs.lift(1).map(_.toInt).getOrElse(4 * 1024 * 1024)
+ val iterations = mainArgs.lift(2).map(_.toInt).getOrElse(5)
+ require(
+ rows >= sourceFiles && iterations > 0,
+ "Need at least eight rows and positive iterations")
if (!icebergOnClasspath) {
// scalastyle:off println
println("Iceberg is not on the classpath; skipping. Build with an Iceberg-enabled profile.")
@@ -295,14 +312,18 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
withTempPath { warehouse =>
configureIcebergHadoopCatalog(warehouse, catalog)
- runBenchmarkWithTable("Iceberg write", 4 * 1024 * 1024) { values =>
+ runBenchmarkWithTable("Iceberg write", rows) { values =>
withTempPath { dir =>
withTempTable("parquetV1Table") {
saveAsParquetV1Table(
spark.sql(corpusQuery).repartition(sourceFiles).write,
dir.getCanonicalPath + "/parquetV1")
try {
- workloads(values).foreach(runWorkload(_, values))
+ if (mode == "all") {
+ workloads(values).foreach(runWorkload(_, values, iterations))
+ } else {
+ runBloomWorkload(values, iterations, mode)
+ }
} finally {
spark.sql(s"DROP TABLE IF EXISTS $targetTable")
}
@@ -312,7 +333,84 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
}
}
- private def runWorkload(workload: Workload, values: Int): Unit = {
+ private val bloomColumns = Set("c_long", "c_str", "c_bin")
+
+ private def runBloomWorkload(values: Int, iterations: Int, mode: String): Unit = {
+ val benchmark = new Benchmark(
+ "Native Iceberg writes, default Bloom cap (1 MiB per column)",
+ values.toLong,
+ minNumIters = iterations,
+ warmupTime = 500.millis,
+ minTime = 500.millis,
+ output = output)
+ Seq(false, true)
+ .filter(enabled => mode == "bloom" || mode == s"bloom-${if (enabled) "on" else "off"}")
+ .foreach { enabled =>
+ val arm = arms.last.copy(
+ name = s"Native Bloom ${if (enabled) "on" else "off"}",
+ // Keep native writes on in BOTH arms. Turning this option off selects Java for
+ // Bloom-enabled tables, which would measure a different writer rather than hashing.
+ confs = arms.last.confs :+
+ (CometConf.COMET_ICEBERG_NATIVE_WRITE_BLOOM_FILTER_FOLDING_ENABLED.key -> "true"))
+ val workload = Workload(
+ title = "unpartitioned INSERT with native Bloom filters",
+ partitionSpec = "",
+ properties = if (enabled) bloomColumns.toSeq.sorted.map { column =>
+ s"'write.parquet.bloom-filter-enabled.column.$column'='true'"
+ }
+ else Nil,
+ orderBy = None,
+ prePopulate = false,
+ statement = s"INSERT INTO $targetTable SELECT * FROM parquetV1Table",
+ expectedRowsAfter = values.toLong,
+ expectShuffle = None,
+ expectSort = None)
+ val label = verifyArm(benchmark, arm, workload)
+ require(label == arm.name, "Bloom benchmark requires native writes in both arms")
+ verifyBloomFilters(if (enabled) bloomColumns else Set.empty)
+ benchmark.addTimerCase(label) { timer =>
+ resetTable(workload)
+ withSQLConf(arm.confs: _*) {
+ timer.startTiming()
+ spark.sql(workload.statement)
+ timer.stopTiming()
+ }
+ }
+ }
+ benchmark.run()
+ }
+
+ private def verifyBloomFilters(expected: Set[String]): Unit = {
+ val files = spark.sql(s"SELECT file_path FROM $targetTable.data_files").collect()
+ require(files.nonEmpty, "Bloom benchmark must produce data files")
+ files.foreach { file =>
+ val input = HadoopInputFile.fromPath(
+ new HadoopPath(file.getString(0)),
+ spark.sparkContext.hadoopConfiguration)
+ val reader = ParquetFileReader.open(input)
+ try {
+ val columns = reader.getFooter.getBlocks.asScala.flatMap(_.getColumns.asScala)
+ require(
+ expected.subsetOf(columns.map(_.getPath.toDotString).toSet),
+ "Benchmark output is missing configured Bloom columns")
+ columns.foreach { column =>
+ val bloom = reader.readBloomFilter(column)
+ require(
+ (bloom != null) == expected.contains(column.getPath.toDotString),
+ s"Unexpected Bloom filter presence for ${column.getPath} in ${file.getString(0)}")
+ if (bloom != null) {
+ require(
+ bloom.getBitsetSize > 0 && bloom.getBitsetSize <= 1024 * 1024,
+ "Bloom filter must fit the default 1 MiB cap")
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ }
+
+ private def runWorkload(workload: Workload, values: Int, iterations: Int): Unit = {
// `warmupTime` is a deadline and `minTime` a total across the measured iterations, so both are
// floors on a case's total cost rather than on one write. The floors are set below the fastest
// write the corpus produces - the native writer is around half a second here - so that every
@@ -326,7 +424,7 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
// its own runtime as spread between iterations, against a few percent for the two Comet arms
// - and it is also the divisor of every `Relative` figure. Five iterations rather than the
// usual three are cheap here and steady the number the table is read for.
- minNumIters = 5,
+ minNumIters = iterations,
warmupTime = 500.millis,
minTime = 500.millis,
output = output)