Repository navigation
feat: support bloom filters in native Iceberg writes - #5724
NikitaMatskevich wants to merge 32 commits into
Conversation
andygrove
added
enhancement
a9254ed to
cd3042f
Compare
alessandro-nori
left a comment
alessandro-nori
left a comment
There was a problem hiding this comment.
found one divergence from iceberg Java, the rest of the implementation looks good to me
cd3042f to
f86e8e4
Compare
f86e8e4 to
1d8376b
Compare
unikdahal
left a comment
unikdahal
left a comment
There was a problem hiding this comment.
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.
1d8376b to
7d218de
Compare
unikdahal
left a comment
unikdahal
left a comment
There was a problem hiding this comment.
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?
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
9875a7b to
fcef2c4
Compare
6cdd0a7 to
ae68947
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed commit ae68947c637ff11f3ce37f7446c58445ff1a93e3. New P1/P2 findings are detailed inline.
| 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) |
There was a problem hiding this comment.
[P2] On Iceberg 1.8.1, tags.a and tags.element.a can name the same leaf of ARRAY<STRUCT<a: INT>>. Enable both and set FPP 0.0001 only on tags.a. Java retains the explicit FPP. Comet emits two entries for tags.list.element.a, and the second resets it to 0.01. Resolve enabled-prefix entries by physical leaf before discarding disabled names or applying defaults. Falling back on collisions also preserves Java’s behavior.
Evidence: The exact-head translator emitted FPPs [0.0001, 0.01]. The tagged 1.8.1 Java setter retained 0.0001. With integers 0..99, the two-entry native file had a 128-byte filter and 9,088 positives among one million absent probes. The single-setting control had 512 bytes and 13 positives. A second fixture with tags.a=true and tags.element.a=false left the Java filter disabled while Comet enabled it.
There was a problem hiding this comment.
Thank you, addressed in #7226, commit fix: fall back for colliding legacy Bloom aliases. This PR keeps these older runtimes on Java.
| getMethod(field.getClass, "fieldId").invoke(field).asInstanceOf[Int] | ||
| } | ||
| } catch { | ||
| case _: Exception => None |
There was a problem hiding this comment.
[P2] On Iceberg 1.8.1, setting write.parquet.bloom-filter-enabled.column. to true makes Java call Schema.findField("") and reject the write. This catch treats the validation exception as a missing field, allowing an opted-in native write to proceed with no filter. Preserve lookup failures as fallback reasons and validate the empty suffix for every enabled-prefix entry, including entries set to false.
Evidence: The tagged Java writer raised IllegalArgumentException: Invalid column name: (empty). The exact-head resolver returned Some(Map()) and serialized zero filters. With enabled=false and an interpreted FPP of 0.01 on the empty suffix, Java also rejected the configuration while Comet selected zero columns. A nonempty missing-name control was skipped by both.
There was a problem hiding this comment.
Thank you, addressed in #7226, commit fix: preserve legacy Bloom lookup validation. This PR keeps these older runtimes on Java.
| val resolvedPaths = version match { | ||
| case Some((1, minor)) if minor < 8 => | ||
| // Iceberg 1.5 passes the configured name straight to parquet-mr. | ||
| configuredColumns.filter(resolution.physicalPaths).map(path => path -> path).toMap |
There was a problem hiding this comment.
[P2] Iceberg 1.5.2 passes the configured spelling directly to parquet-mr. For an id INT column, write.parquet.bloom-filter-enabled.column.id.=true therefore enables the id filter because ColumnPath.fromDotString drops the trailing empty component. The exact-string lookup here drops the setting, so the native file loses the requested filter. Normalize raw paths with parquet-mr’s rules before matching them, or fall back for these spellings.
Evidence: A parquet-mr 1.13.1 writer produced a 1,048,576-byte filter on id for both id. and id. The exact-head 1.5.2 resolution branch returned an empty map for id. and Map(id -> id) for the control.
There was a problem hiding this comment.
Thank you, addressed in #7226, commit fix: normalize legacy Iceberg Bloom paths. This PR keeps these older runtimes on Java.
| collectWithSubqueries(plan) { case exec: CometIcebergWriteExec => exec } | ||
| } | ||
| assert( | ||
| nativeExecs.isEmpty, |
There was a problem hiding this comment.
[P2] This assertion expects JVM fallback on Spark 3.4/Iceberg 1.5.2, but the fixture configures logical order id. That runtime uses raw Parquet names, so this spelling matches no physical leaf and the planner permits native execution. Configure order_x20id on 1.5.2 to exercise the intended fallback and retain the logical spelling on runtimes that resolve it.
Evidence: The 1.5.2 resolution branch returned Some(Map()) for order id and selected fallback for order_x20id. Parquet 1.13.1 footers showed zero filter bytes for the logical spelling and 1,048,576 for the physical spelling. This verifies the assertion mismatch at the component level. The full Spark 3.4 suite stopped before test execution.
There was a problem hiding this comment.
Thank you, addressed in #7226, commit test: use physical Bloom names on Iceberg 1.5. This PR keeps these older runtimes on Java.
|
I decided to split this into modern-runtime support here (identifiable Iceberg 1.10+) and older Spark/Iceberg support in #7226. The follow-up branch is based on this PR and restores legacy support with the naming and validation fixes discussed in review. This PR documents Java fallback below Iceberg 1.10, including in the compatibility matrix. I am also OK combining the two PRs into one if preferred. |
|
Added a Benchmark results section to the PR description, following Andy’s suggestion: native Bloom off/on write time and whole-process peak RSS at the default cap. |
|
A possible simplification after a dependency upgrade:
Once Comet adopts compatible dependencies containing all 3 improvements, it could set them internally removing the reason for a specific Comet opt-in flag. Byte identity would apply to filters with matching physical values, allocation, and column-chunk contents; it would not guarantee identical whole-file or row-group layouts. This is a follow-up direction; the current PR retains its explicit opt-in. |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for the work on this, and for splitting the legacy runtimes out. The sizing mirror of BlockSplitBloomFilter, the physical-path resolution, the float and double fallback and the folding opt-in all look right to me. I also went back through the earlier threads and I think the earlier findings from sunchao, jordepic and my own passes are addressed, either in this PR or by sending those runtimes to Java.
I compared the gate against branch-1.2. There, write.parquet.bloom-filter-enabled.column.*=false is vetted and stays native, so this PR changes behavior for a table that sets it on Iceberg older than 1.10. I left that as an inline comment because it is the one thing I would like changed before merge. Everything else I found differs from the release in an intended way: enabled=true with the opt-in on, FPP and NDV on an enabled column, and a positive unused max-bytes.
CI has not run on any revision of this branch because the fork workflows were waiting for approval. I approved the Comet CI and CodeQL runs on 9c468b65 and added run-iceberg-tests next to the existing run-all-spark-profiles, since this changes the Iceberg write gate. I will look at the results when they finish. Nobody has run the Spark 4.0 / Iceberg 1.10 profile yet, so that one will be the first real signal.
One question on #7226. It is based on this branch and the legacy behavior depends on both PRs. Would you be open to combining them, as you offered? That would keep one review of the gate and avoid a window where main is slower for older Iceberg runtimes.
| } | ||
| } | ||
|
|
||
| private val requireSupportedBloomFilterRuntime: TriggerRule = ctx => { |
There was a problem hiding this comment.
On Iceberg 1.5 and 1.8 this rule rejects as soon as any key with a Bloom prefix is present, including write.parquet.bloom-filter-enabled.column.id=false. In branch-1.2 that property is vetted and the write stays native, so a table that only sets enabled=false on Spark 3.4 or 3.5 goes from native to Java with this change. legacy runtime rejects every configured Bloom entry before filtering disabled names and legacy Bloom settings use Java despite the native folding opt-in pin that behavior. A property that writes no filter has nothing for the runtime check to protect, and the empty-suffix and unknown-column keys are skipped by every runtime anyway.
Could the runtime check apply only when hasEnabledBloomFilters(interpretedBloomFilterProperties(ctx.properties)) is true? Then enabled=false and unmatched entries keep their current native behavior on older runtimes, and standalone FPP and NDV keys still fall back through the unvetted-key rule there. If you would rather keep this PR narrow and let #7226 restore it, the two PRs need to land together so no release ships the regression.
comphead
left a comment
comphead
left a comment
There was a problem hiding this comment.
I read the whole thread and did not repeat its settled points (physical paths, NDV and cap handling, capability gating per runtime, sanitized names, the folding opt-in). Iceberg 1.10.0 through 1.12.0 resolve Bloom columns the same way, the sizing mirror matches parquet 59.3.1 and parquet-mr, and andygrove's latest inline point on enabled=false for legacy runtimes is still open. Three comments.
| builder = builder | ||
| .set_column_bloom_filter_enabled(path.clone(), true) | ||
| .set_column_bloom_filter_fpp(path.clone(), fpp) | ||
| .set_column_bloom_filter_max_ndv(path, synthetic_ndv); |
There was a problem hiding this comment.
Since #6247, each open file's share of the reservation comes from MeteredParquetWriter, and its doc comment says the share would not see Bloom filters, "which parquet-rs sizes when a file opens and the eligibility gate declines today". This PR lifts that gate without adding the filters to the share. In parquet 59.3.1, create_bloom_filter calls Sbbf::new_with_ndv_fpp when each row group's column writer is created, and new_with_num_of_bytes allocates the whole bitset at once, so with the synthetic NDV every Bloom column holds its full target, 1 MiB by default and up to 128 MiB, from the first row. A fanout write holds that for every Bloom column of every open partition file, none of it reaches try_resize, and by_memory() cannot pick the files that hold it. Could each file's share include the sum of its Bloom targets? It is fixed for the file's lifetime, since the default AfterRowGroup position drops each row group's filters once written. The doc comment and the user guide's memory paragraph would change with it. I derived this from the source and have not measured it.
| 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 |
There was a problem hiding this comment.
One more file-level difference could go in this list. build_writer_properties keeps parquet-rs's default BloomFilterPosition::AfterRowGroup, so each row group's filters follow that row group, while parquet-java writes every filter in ParquetFileWriter.end() after the last row group. Readers find filters through bloom_filter_offset, so I do not see one that could observe it, but listing it with that reason would match the other entries.
| pathByFieldId: Map[Int, String], | ||
| physicalPaths: Set[String], |
There was a problem hiding this comment.
pathByFieldId and physicalPaths are built, but I could not find a reader of either in this PR. They are used only by the 1.5 to 1.9 branches in #7226. If the two PRs stay separate, could these fields move there?
Which issue does this PR close?
Partially addresses #5643 and #5649. Does not close the generic Parquet writer issue #5304.
Rationale for this change
Enable native Iceberg writes for tables with column Bloom filters on identifiable Iceberg 1.10+ runtimes. Older or unidentified runtimes use Java whenever any per-column Bloom property is configured, including false enablement and standalone FPP/NDV. Tables without those properties retain their existing native-write eligibility.
The default-off
spark.comet.write.iceberg.bloomFilterFolding.enabledoption accepts parquet-rs's folding behavior. Sparse filters can shrink toward the requested FPP, trading serialized space for potentially weaker pruning than Java's larger filters. The flag selects the writer; it cannot disable folding in the pinned parquet-rs implementation.We are also pursuing an upstream parquet-rs option to disable folding, which would simplify compatibility with Java by retaining the initial filter size. The current Comet opt-in accepts folding; it does not disable it.
What changes are included in this PR?
How are these changes tested?
Benchmark results
Following Andy Grove’s suggestion, the existing
CometIcebergWriteBenchmarkcompares native writes with Bloom filters off versus on at the default 1 MiB cap per column. Bloom filters are enabled on the bigint, string, and binary columns.Release native build; Spark 4.1 / Iceberg 1.11; JDK 21.0.6; Apple M4 Max, 64 GiB RAM;
local[5], 8 GiB JVM heap. Each case writes 4,194,304 rows, with five measured iterations in a separate JVM.Bloom filters added approximately 6% to average write time in this local run. Peak RSS includes Maven/JVM startup, source generation, validation, and warmup; the lower observed RSS with filters enabled does not establish a memory saving. These measurements compare the native writer with filters off/on; they do not measure Java/native writer speed or reader pruning.