Repository navigation
feat: support bloom filters in native Iceberg writes - #5724
NikitaMatskevich wants to merge 31 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. |
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.