feat: support bloom filters in native Iceberg writes

[NikitaMatskevich]

Which issue does this PR close?

Partially addresses #5643 by lifting the native Iceberg V2 write restriction for the complete supported Bloom-filter property set:

• write.parquet.bloom-filter-enabled.column.<column>
• write.parquet.bloom-filter-fpp.column.<column>
• write.parquet.bloom-filter-ndv.column.<column>
• write.parquet.bloom-filter-max-bytes, when its value is exactly representable by the Apache Arrow Rust parquet crate

This is also part of the production-quality native Iceberg write work tracked by #5649.

This is related to #5304, which tracks missing writer-property propagation in the generic native ParquetWriterExec, but #5304 is not a blocker for this PR and is not closed by it. Iceberg V2 writes use a separate native writer path and construct their own WriterProperties.

Why is this change needed?

Before this PR, an Iceberg table that enabled a Parquet Bloom filter could not use Comet's native Iceberg writer.

Bloom-filter enablement alone is not enough for production compatibility. Iceberg exposes the requested false-positive probability (FPP), expected number of distinct values (NDV), and maximum Bloom-filter allocation. Ignoring any of these properties could silently produce a differently sized filter and regress pruning for readers.

This PR therefore translates the complete sizing decision, uses the native writer only when the result can be represented without any regression, and otherwise keeps the classic writer fallback.

Writer architecture

The write path is:

Spark Iceberg V2 write
└── JVM `IcebergWriteExec` (Comet split writer / JVM fallback)
    └── JVM `CometIcebergWriteExec` (native-execution wrapper)
        └── protobuf `Operator { iceberg_write: IcebergWrite }`
            └── Rust `execution::operators::iceberg_write::IcebergWriteExec`
                ├── Comet-built `parquet::file::properties::WriterProperties`
                └── iceberg-rust `ParquetWriterBuilder::new(WriterProperties, schema)`

It constructs WriterProperties from the Apache Arrow Rust parquet crate
and passes them into iceberg-rust: dee native/core/src/execution/operators/iceberg_write.rs.
That direct construction is useful: Comet does not have to delegate Iceberg table-property translation to iceberg-rust's from_table_properties. The pinned Apache Arrow Rust parquet crate 58.4.0 API already exposes set_column_bloom_filter_enabled, set_column_bloom_filter_fpp, and set_column_bloom_filter_ndv, so no iceberg-rust contribution is required for this work.

Sizing compatibility strategy

Iceberg's defaults are FPP 0.01 and max-bytes 1 MiB. The JVM Iceberg/Apache Parquet Java stack applies the settings in this order:

1. If NDV is absent, request the full max-bytes allocation.
2. If NDV is present, calculate the requested allocation from NDV and FPP.
3. Cap that allocation at max-bytes.

Comet reproduces this precedence.

Apache Parquet Java accepts a non-power-of-two cap and can serialize that exact used space, while the Rust crate uses power-of-two allocations and may fold a sparse filter after writing. For that reason, non-power-of-two, malformed, and out-of-range values fallback to classic writer.

The Apache Arrow Rust parquet crate 58.4.0 has no independent max-byte setter. Comet first reproduces Apache Parquet Java's allocation decision, then converts the resulting power-of-two byte target into a synthetic NDV passed to the Rust parquet crate. For a target B, the crate rounds every calculated size in (B/2, B] up to B; Comet aims at 3B/4, verifies the result using the crate's exact sizing expression, and has a binary-search fallback for unusual valid FPP values. This avoids fragile +/- 1 behavior at floating-point boundaries.

Folding: the Apache Arrow Rust parquet crate may fold a sparsely populated filter to a smaller power-of-two filter while preserving the configured FPP. This is considered an accepted improvement: it reduces file space and remains safe for every standards-compliant Parquet reader. Where folding does not occur, this PR verifies byte identity with Apache Parquet Java. Where it does occur, tests compare both used space and membership safety explicitly.

Feature	Supported by this PR	Upstream work required?
Per-column Bloom-filter enablement	Yes	No
Per-column FPP configuration	Yes	No
Per-column NDV configuration	Yes	No
Power-of-two write.parquet.bloom-filter-max-bytes in [32, 128 MiB]	Yes	No; Comet emulates it through a verified synthetic NDV
Arbitrary non-power-of-two max-byte parity	No; classic-writer fallback	A future Apache Arrow Rust parquet API would still need exact arbitrary-size and folding semantics before it could remove this fallback safely
Automatic Iceberg property translation in iceberg-rust	Not used	No; direct WriterProperties construction is intentional

What changes are included?

• The JVM driver resolves enabled Iceberg logical column names to physical Parquet leaf paths
• The native writer mirrors Apache Parquet Java's allocation decision and configures the Apache Arrow Rust parquet crate per column.

Commit structure

This PR is intentionally split into two reviewable feature commits:

1. Enable per-column Bloom filters in native Iceberg writes (d60c19f). This commit plumbs write.parquet.bloom-filter-enabled.column.<column>
2. Honor Iceberg Bloom-filter sizing settings (b544d41). This commit adds FPP, NDV, max-byte, and nested physical-path translation

Each commit is a self-contained feature increment with its own implementation and tests. Both feature states were validated with their relevant CI test suites locally and can be split into distinct PRs if reviewers prefer to review or merge enablement and sizing separately.

End-to-end writer and file tests: CometIcebergWriteActionSuite

• Configured Bloom filters exist and native execution is real.
• Binding max-bytes byte identity.
• FPP/NDV presence matrix and impossible settings.
• Underestimated explicit NDV.
• Large cap, folding, byte identity, and adjacent fallback.
• Out-of-range writer selection.

Planning tests: CometIcebergWriteDetectionSuite

JVM/protobuf tests: IcebergWriteProtoTranslationSuite

• Proves the unset defaults are exactly FPP 0.01 and max bytes 1,048,576.
• Proves explicit per-column FPP/NDV and explicit max bytes are transferred deterministically.
• Proves logical Iceberg list/map leaf names are remapped to physical Parquet paths, while properties for missing fields are skipped like Iceberg Java.

Rust unit tests: execution::operators::iceberg_write::tests

Assisted by Codex.

[alessandro-nori]

found one divergence from iceberg Java, the rest of the implementation looks good to me

[alessandro-nori]

this differs from the Java implementation for map and list columns (e.g. for a list tags.element vs tags.list.element) and the filter would be silently omitted by Java readers.

In iceberg-rust there is a schema visitor called IndexByParquetPathName but it is private.
Could we consider resolving the Parquet paths on the driver (Scala)? Or maybe making the iceberg-rust visitor public

[NikitaMatskevich]

Thanks for reviewing it! Implemented and pushed in the amended second commit 1d8376b. Test covering this was added too

[unikdahal]

Thanks for the detailed work here. The nested list/map physical-path fix looks good.

I found three remaining parquet-mr compatibility gaps in Bloom-filter property handling/sizing. Details inline.

[unikdahal]

I don't think we should intentionally change the max-bytes=32 behavior while describing this sizing logic as parquet-mr compatible.

In parquet-mr, BlockSplitBloomFilter only installs maximumBytes when it is strictly greater than the 32-byte lower bound. So a configured maximum of exactly 32 is effectively not used as the maximum when NDV/FPP request a larger filter.

For example:

NDV=1,000,000, FPP=0.0001, max-bytes=32

requests about 2.63 MiB before power-of-two rounding, and the JVM writer ends up with a 4 MiB Bloom filter because the 32-byte maximum is ignored. This implementation forcibly returns 32 bytes instead — a very large pruning-quality difference.

Could we either emulate the parquet-mr behavior here or conservatively fall back to the classic writer for max-bytes=32 when it matters? The existing binding-cap test would be stronger if it compared the JVM and native footer sizes for this boundary.

[NikitaMatskevich]

Ok, changed this behavior to comply with parquet-mr in 7d218de.

[unikdahal]

There is one more parquet-mr parity edge case here for very large but still valid NDVs.

parquet-mr calculates -8 * n using Java long arithmetic before the division/conversion to floating point. That multiplication can overflow. Here ndv is converted to f64 before multiplication, so the Java overflow behavior can never occur.

For example, with NDV = 2^61, Java's -8 * n wraps to zero and parquet-mr ends up requesting the minimum 32-byte Bloom filter, while this implementation calculates a huge value and caps it at the configured maximum (1 MiB by default).

Since planning currently accepts the full positive Java long range, this can silently select native execution with materially different output. Rather than reproducing the overflow, could we conservatively fall back for ndv > Long.MaxValue / 8 and add tests at the threshold, threshold + 1, 2^61, and Long.MaxValue?

[NikitaMatskevich]

This does not look like a feature one can actually rely upon. But I guess it doesn't hurt to have this behavior replicated here. Merged in 7d218de.

IMO so many distinct values in a column is not a reasonable usecase for bloom filters anyway, so producing a tiny 32b "placeholder" filter for such columns is even better implementation choice than actually allocating 128mb. That said, I would even lower this "NDV max limit" down from 2^60 to something more realistic if I could.

[unikdahal]

Thanks for addressing the earlier comments. I re-checked the latest changes, and those issues look resolved.

I found one remaining cross-version test issue and left an inline comment on it. Other than that, the implementation LGTM.

Also, could you please rebase against main to resolve the merge conflicts?

[unikdahal]

Could we gate NDV support separately here? This only checks the FPP property, but NDV is absent on older supported Iceberg versions (e.g. 1.8/1.10). The assumption therefore passes while production correctly falls back for explicit NDV, so NDV tests such as the overflow / false + NDV cases can incorrectly expect Compatible.

Separate FPP and NDV assumptions would also preserve FPP-only coverage on those versions.

[sunchao]

Correctness

Reviewed 7d218de7 against authoritative base 7f1e0018, with authored changes measured from merge base 719cba11. Previously, an enabled Bloom-filter property forced Iceberg writes onto the JVM path. The PR lifts that restriction, maps configured columns to Parquet paths, carries FPP/NDV/max-byte settings through protobuf, and configures the native writer. Its eligibility checks keep unsupported runtime properties, unrepresentable caps and problematic numeric values on the classic writer.

The previous discussions about enabled=false plus NDV, the 32-byte cap and overflowing NDV are addressed in the current control flow. NDV can re-enable a configured column, associated malformed values are checked even when enabled is false, and the two sizing boundaries fall back. The list/map path adjustment also addresses the previously reported missing intermediate path components.

One new [P2] remains: the new map assumes Iceberg Java's physical field names are also the native writer's names. Java sanitizes names such as order id to order_x20id, while the pinned native schema conversion preserves order id. The native Bloom-property lookup therefore misses the configured column. The inline comment requests matching the native writer's schema or falling back for renamed fields.

The maintained Spark 3.5 and 4.0 parsers accept backquoted names containing spaces and punctuation, and their schemas retain the names. This is a valid column shape, not an invalid-name input. Spark's V2 commit/abort protocol is unchanged by this PR. Bloom filters affect pruning rather than row values, so correct row results alone cannot establish that these properties were honored.

The existing cross-version test issue also remains. Both suites' shared assumption checks only the FPP constant. Iceberg 1.8.1 and 1.10.0 expose that constant but lack NDV support, so the assumption passes while production correctly rejects explicit NDV. NDV cases that expect native execution or Compatible need a separate capability assumption. I have not duplicated that inline comment.

Validation

No check runs, statuses or Actions runs were present for this head in the fresh check at 2026-09-08 04:40:59 UTC. No CI checkout, merge-tree equivalence or native artifact provenance could therefore be credited. The author's local-suite statement is not independently verified. The assigned base has advanced beyond this branch's merge base, including overlapping writer/reflection/test changes that must survive the requested rebase.

A bounded component test compiled the exact Java name-sanitizing methods, parquet-rs column-path identity/conversion and Comet path/sizing functions. Plain names matched. Three sanitized names missed the native column keys, and controls using native paths matched. It also passed 115 allocation round trips across supported powers of two and representative FPPs, plus default, underestimated-NDV, binding-cap and pathological-FPP controls. This used a simple property-map fixture and a Java precondition stub. It was not a complete Parquet-file or Spark/JNI reproduction. No local Comet build, Spark suite or performance benchmark was run. Maintained Spark 3.4/4.1 source branches were unavailable.

Performance

The default requests up to 1 MiB per configured column per row group, with larger supported caps up to 128 MiB. Allocation and hashing occur in the native writer, and fanout can multiply concurrent allocations. The sizing translation preserves the intended cap/NDV precedence in the inspected cases. The pinned parquet-rs encoder folds filters on flush, so final file size can differ from initial allocation.

The new path mismatch silently loses requested Bloom pruning for affected names. Apart from that finding, I found no material unnecessary work introduced by the translation. Schema-path reflection is performed during driver serialization, and the inverse sizing usually takes one candidate with a bounded search fallback. File-size assertions and sizing calculations do not establish write-throughput or downstream query gains.

Design

Separating eligibility, protobuf translation and native writer-property construction keeps fallback decisions before task execution. Explicit defaults prevent parquet-rs's different FPP default from leaking into Iceberg writes, and preserving absent NDV is necessary for the JVM sizing precedence. The physical-path mapping is the weak boundary: its source must agree with the schema actually emitted by the native writer.

Abstraction & complexity

The synthetic NDV is justified by the pinned API's lack of a separate byte-cap setter. The inverse calculation verifies its result, and the numeric search is bounded. The JVM representability check duplicates part of that sizing logic, but it serves a distinct purpose by rejecting unsupported settings before execution. Keeping those checks tied to dependency versions and boundary tests is necessary. Beyond correcting the path mapping, I found no actionable abstraction change.

[sunchao]

Correctness

[P2] Resolve Bloom paths against the native writer schema

Could we account for field-name sanitization before using this Java schema as the native path map? Iceberg Java converts a quoted column such as order id to physical name order_x20id, but the pinned iceberg-rust ToArrowSchemaConverter and parquet-rs writer preserve order id. With write.parquet.bloom-filter-enabled.column.order id=true, this map therefore configures ColumnPath(["order_x20id"]) while the writer looks up ["order id"]. The native write silently omits the requested Bloom filter. The same mismatch affects dotted names and names starting with a digit.

Please derive paths from the schema the native writer actually emits, or fall back when the Java conversion renames a configured field, and add a quoted-name test that checks the written footer for the filter. A bounded component using the exact name/path methods reproduced the lookup miss for three such names, with plain-name and native-path controls passing. This was not a full Spark/JNI reproduction.

[NikitaMatskevich]

Thank you for review! Fixed in "fix: fall back for sanitized Iceberg bloom paths" by falling back to java implementation when those problematic column names occur. I don't think its a big blocker for the coverage of Comet, personally I didn't see such naming in production before.

[NikitaMatskevich]

Added 2 boolean flags icebergSupportsBloomFpp and icebergSupportsBloomNdv based on reflection of Fpp and Ndv string prefixes availability in current iceberg-java. Based on these flags, now tests verify expected behaviors for each scenario. This was pushed in commit "fix: match Iceberg-version bloom property support".

• Spark 4.1 / Iceberg 1.11: complete writer suites, 115/115 passed; final parity test also rerun successfully.
• Spark 3.4 / Iceberg 1.5.2: detection and action suites passed after correcting the version-aware footer assertion.
• Spark 3.5 / Iceberg 1.8.1: full suites passed; final FPP-only test rerun successfully.
• Spark 4.0 / Iceberg 1.10.0: full suites passed; final FPP-only test rerun successfully.

[sunchao]

Correctness

Follow-up at fcef2c4f against bb9e7402, after the review of 7d218de7. The earlier sanitized-name finding is addressed: physical-path resolution compares each named ancestor and leaf with its Iceberg field ID, and an enabled column requiring Java name sanitization causes JVM fallback. That covers literal dots as well as spaces in field names by source inspection. The new action test specifically exercises order id. Maintained Spark 3.5/4.0 accept quoted identifiers, so preserving the JVM write path for these names is necessary. Their DataSource V2 commit/abort contract is unchanged by this update.

The FPP/NDV capability finding is also addressed. Iceberg 1.5.2 ignores both shape properties, 1.8.1/1.10.0 interpret FPP, and 1.11.0 additionally interprets NDV. Filtering unsupported prefixes before validation and serialization now preserves those versions' behavior, including enabled=false, malformed ignored properties, and the NDV setter re-enabling a filter when supported. The revised tests gate FPP independently, retaining the older-runtime FPP-only case. The positive-long/overflow, finite-FPP and representable-cap fallback checks remain in place. This change does not alter value conversion, null handling, or ANSI/Legacy expression semantics.

One new P2: the Bloom code still uses the pre-59 Parquet API after the rebase. The new test accesses private BloomFilterProperties.fpp/ndv fields, and the production setter is deprecated under the CI warnings-as-errors policy. The inline comment identifies both required API updates.

Validation

The exact dependency API projected into separate Rust crates reproduces three private-field errors and the deprecated-setter error. Getter/current-setter controls compile. The extracted sizing helper passes 115 allocation round trips and default, binding-cap, underestimated-NDV and extreme-FPP controls. These are component checks, not a full Comet/JNI/Spark build. Diff and Rust format checks pass. At the September 8, 10:41 UTC refresh, all four current-head workflows remain action_required, with zero check results. The recorded workflow-job inventories contain zero jobs. The author's reported Spark/Iceberg suite results are separate evidence, not independently reproduced current-head qualification. Maintained Spark 3.4/4.1 source coverage remains unavailable.

Performance

The new path traversal and capability filtering run during planning. No new per-row reflection is introduced. Writing enabled filters still adds hashing and allocates the initial configured filter per column and open row group. Parquet 59.3 folds at flush using an in-place truncation that retains vector capacity, so smaller serialized filters do not establish a corresponding memory reduction.

Please add a focused native/JVM write benchmark with Bloom disabled and enabled, including the default cap and a large cap at low and estimated cardinalities. Report write time, peak memory and filter/file bytes, then measure equality/IN row groups and bytes read with an identified Bloom-aware reader. Iceberg Java's reader has that pruning path. The inspected pinned iceberg-rust scan pipeline uses metrics/page/row filtering without loading Bloom filters. Footer byte equality and membership checks establish neither end-to-end speedup nor native-reader pruning benefit.

Design

The renamed-column fallback is an appropriate bounded fix while the bridge uses dot-separated physical paths. Comparing ancestors prevents a renamed parent from escaping a leaf-only check. Centralizing runtime capability filtering keeps eligibility and serialization consistent, while retaining the explicit JVM fallback for sizing that cannot be represented. No additional design blocker was found in the follow-up.

Abstraction & complexity

The path-resolution result earns its two fields by carrying the mapping and the fallback decision together. The synthetic-NDV adapter remains relevant because the current Parquet writer configuration sizes from NDV/FPP rather than accepting the desired byte cap directly. Its round-trip check makes that boundary explicit. Update its API calls for Parquet 59, but no further abstraction is needed for these fixes.

[sunchao]

Correctness

[P2] Use the Parquet 59 Bloom API after the rebase

The current lockfile resolves parquet 59.3.0, where BloomFilterProperties.fpp and .ndv are private. These three field accesses therefore fail with E0616 when the native tests are compiled. Use bloom.fpp() and bloom.ndv(). Also update the new setter at line 685 from set_column_bloom_filter_ndv to set_column_bloom_filter_max_ndv: the old name is deprecated since 59.0 and fails the Rust CI action's -- -D warnings check. Separate-crate projections of the exact upstream API reproduce both failures, while getter/current-setter controls compile. Please run the native Rust checks after both updates.

[NikitaMatskevich]

My bad, Codex did not re-run the tests after rebase, the fix was merged here: fix: use Parquet 59 bloom filter APIs

[sunchao]

Correctness

The Parquet 59 API finding from the previous review is addressed at 709a3604. Production now calls set_column_bloom_filter_max_ndv, and the test uses fpp() and ndv(). In the exact locked Parquet 59.3.0 source, the old setter delegates to this method and the getters return the same stored values. The update preserves sizing behavior while removing both the deprecated call and private-field accesses. No new or remaining verified P1/P2 was found.

The incremental diff from fcef2c4f contains only those three changed lines. The authoritative base remains bb9e7402. The other eight authored files, lockfile and Rust CI policy are unchanged. The earlier sanitized-name fallback and independent FPP/NDV capability fixes remain intact, including the FPP-only test path. Maintained Spark 3.5/4.0 quoted-name and V2 write semantics remain the compatibility reference. This API adjustment does not change column names, null/value handling, errors, expression modes or commit/abort behavior.

Validation

Diff and Rust format checks pass. The exact API source and preserved positive controls support the fix, but no new local native/JNI/Spark execution is credited. At the September 8, 12:26 UTC refresh, all three current-head workflows are action_required. Their subsequent job inventories contain zero jobs, including the CI run. The reported merge has the assigned head/base as parents and exactly the head tree, but it has not supplied an executed Rust check. The author's latest reply acknowledges that tests were not rerun after the earlier rebase and links this fix. Maintained Spark 3.4/4.1 source coverage remains unavailable.

Performance

No new timing, peak-memory or reader-pruning results were supplied. The API replacement adds no work beyond the operation already performed by the deprecated alias. The previous performance qualification remains open: folded serialized size does not establish lower retained allocation, writer speedup or native-reader pruning. The requested matched native/JVM enabled/disabled writer and Bloom-aware reader measurements remain the evidence needed for those claims.

Design

Calling the supported dependency API directly is the simplest fix. The existing eligibility, path mapping and JVM fallback boundaries are unchanged. No additional design concern was found in this update.

Abstraction & complexity

The change adds no wrapper, compatibility shim or new abstraction. Keeping the dependency-specific API usage inside the existing writer-property construction and its test preserves the current structure without adding indirection.

[sunchao]

Correctness

No new or remaining verified P1/P2 at 298cdc76. Relative to the previous approval on 709a3604, this update only adjusts Markdown table padding and removes two unnecessary Scala s prefixes. Both literals contain neither interpolation nor escape sequences, so the fallback message and generated SQL are unchanged. The authoritative base remains bb9e7402.

The six other authored files, native dependency lock and Rust CI policy are byte-identical to the previously reviewed revision. The Parquet 59 public setter/getter fix, renamed-path fallback, independent FPP/NDV capability gates and regression coverage remain intact. This update preserves the previously checked Spark 3.5/4.0 write behavior.

Validation

Authored, base and incremental diff checks and the Rust format check pass. All discussion was reread and reconciled with the supplied 2026-09-08 14:19:11 UTC snapshot. The only new discussion is the previous approval. There are no new author test results. At the September 8, 14:26 UTC refresh, all three current-head workflows are action_required, and subsequent inventories show zero jobs, including the CI run. The reported merge has the assigned parents and the head tree, but supplies no executed test result. No current native/JNI/Spark execution is credited. Maintained Spark 3.4/4.1 source coverage remains unavailable.

Performance

There is no new benchmark evidence or change to writer, sizing or reader logic. The earlier measurement requests remain outstanding: matched native/JVM writes with filters enabled and disabled, peak and retained allocation, serialized filter bytes, and equality/IN pruning measured through an identified Bloom-aware reader. Folding retains vector capacity, so smaller serialized filters alone establish neither lower retained memory nor a native-reader speedup. This formatting follow-up adds no measured performance claim.

Design

The edit keeps the existing eligibility and JVM fallback boundaries intact. Removing interpolation from fixed text is appropriate and leaves the adjacent SQL interpolation for the table and NDV unchanged. No further design issue was found.

Abstraction & complexity

No helper, compatibility layer or new abstraction is introduced. The two literals become simpler without changing the surrounding control flow or test scope.

[NikitaMatskevich]

Thank you for approval! We synced with @alessandro-nori and his approval is not required. If the CI is green we can merge