Repository navigation
feat: support native Iceberg Bloom writes on older runtimes - #7226
NikitaMatskevich wants to merge 40 commits into
Conversation
andygrove
left a comment
There was a problem hiding this comment.
Thanks for extending this to the older runtimes. I checked the version buckets against Iceberg's Parquet.java at each tag. 1.6.1 and 1.7.2 both add the independent FPP loop, so a garbage FPP throws even for a column that is not enabled. 1.8.1 and 1.9.2 resolve names by field ID with the same code, apart from a null-ID filter. That matches the gate, the docs and the unit tests. I also looked at the opt-in default and the release branch, and I don't see a behavior change for anyone who has not enabled bloomFilterFolding.
This branch is stacked on #5724 and shows 39 commits and about 2,900 added lines against main, so the real change is only visible through the compare link. Could you mark it as a draft until #5724 merges, and then rebase so the diff contains only the legacy changes? Any change made to #5724 in review will also conflict here.
The new tests that are specific to 1.6 and 1.7 skip on every pinned profile, so your local 1.7.2 run is the only evidence for them. Could you add the command you used for that run to the Iceberg test notes in docs/source/contributor-guide/iceberg-writes.md, so someone else can repeat it?
I approved the pending workflow runs and added run-all-spark-profiles, since the legacy paths only execute on the Spark 3.4 and 3.5 profiles. I'll look again once that run reports.
| | `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.<col>` | `true` or `false`; native eligibility requires identifiable Iceberg 1.5+. Unknown runtimes and unsupported legacy settings use Java. 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.<col>` / `write.parquet.bloom-filter-ndv.column.<col>` | Iceberg 1.5 ignores FPP and NDV; 1.8–1.10 interpret FPP but ignore NDV; 1.11 interprets both. Explicit FPP on 1.6/1.7 uses Java. 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`. | |
There was a problem hiding this comment.
This row says what 1.5, 1.8 to 1.10 and 1.11 do with NDV, but not 1.6 and 1.7. Both of those ignore NDV and only interpret FPP through the independent validation pass. Could you say so here, so the row covers every version in the runtime table below?
comphead
left a comment
There was a problem hiding this comment.
Reviewed against #5724's head. The version buckets match Iceberg's Parquet.java at 1.5.2, 1.6.1, 1.7.2, 1.8.1 and 1.9.2, the raw-path normalization matches ColumnPath.fromDotString, and the collision fallback covers the 1.8 and 1.9 field-id resolution. andygrove's draft and test-command requests are not repeated. Two comments.
| val resolution = IcebergReflection | ||
| .getParquetPathResolution(schema) | ||
| .getOrElse(throw new IllegalStateException("Could not resolve Bloom Parquet paths")) | ||
| val _ = bloomFilterParquetPaths(columns, resolution, schema = schema) |
There was a problem hiding this comment.
This rule always returns None and declines only by letting bloomFilterParquetPaths throw from require. getSupportLevel's NonFatal handler then reports "Iceberg native write detection failed: requirement failed: Multiple Bloom column names resolve to the same Parquet leaf: ...", so an intended decline reads like a reflection failure. Could bloomFilterParquetPaths return Either[String, Map[String, String]], so the rule returns the collision or empty-name reason itself? requireNoFloatingPointBloomFilters resolves the schema and paths again, and convert a third time, so folding the FLOAT/DOUBLE check into this rule would resolve once in the gate.
| } | ||
| } | ||
|
|
||
| private def configuredBloomColumns(properties: Map[String, String]): Seq[String] = |
There was a problem hiding this comment.
This is the same enumeration as the inline configured in requireNativeSupportedBloomFilterProperties and the private IcebergWriteProtoTranslation.configuredBloomFilterColumnNames, apart from .sorted in the last. Could the translation's version become private[operator] and replace both?
Which issue does this PR close?
Follow-up to #5724; contributes to #5643 and #5649.
Rationale for this change
Expand #5724's native Bloom-write feature to older Spark/Iceberg runtimes. See that PR for feature details. This branch is stacked on its modern-only head.
What changes are included in this PR?
Legacy name resolution and validation, safe Java fallback for ambiguous settings, and expanded eligibility documentation. Review the legacy-only diff: NikitaMatskevich/datafusion-comet@4b0878d...feat/iceberg-bloom-filter-writes-legacy
How are these changes tested?
Spark 3.4 / Iceberg 1.5.2: 239 tests passed; Spark 3.5 / Iceberg 1.8.1: 253 passed. Modern translation, detection, and Bloom regressions pass on Spark 4.0/4.1. An additional Iceberg 1.7.2 FPP-validation run passed 33 tests. Expected runtime-specific cancellations remain; each fix has focused coverage.