NikitaMatskevich opened a new pull request, #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724
## Which issue does this PR close?
Partially addresses #5643 by lifting the native Iceberg write restriction
for:
`write.parquet.bloom-filter-enabled.column.<column>=true`
This is also part of the production-quality native Iceberg write work
tracked by #5649.
Related to #5304, which tracks missing writer-property propagation in the
generic native `ParquetWriterExec`. That issue is not closed by this PR because
Iceberg
writes use a separate native writer path.
## Rationale for this change
Native Iceberg writes currently fall back to iceberg-java whenever a table
enables a Parquet bloom filter for an individual column.
This restriction was introduced as part of the conservative native Iceberg
write eligibility allowlist in #5298. The native Iceberg writer subsequently
landed in
#5361, but bloom-filter table properties remained unsupported by the
eligibility gate.
As documented in #5643, bloom filters are likely to occur on production
Iceberg tables. Falling back because of this property prevents an otherwise
supported
Iceberg ETL write from running through Comet, even though the underlying
parquet-rs writer already supports producing per-column bloom filters.
Before this change, a table such as:
```sql
CREATE TABLE catalog.db.events (
id INT,
payload STRING
)
USING iceberg
TBLPROPERTIES (
'write.parquet.bloom-filter-enabled.column.id' = 'true'
)
would use the iceberg-java writer instead of CometIcebergWriteExec.
This PR forwards the enabled column names to the native writer, configures
parquet-rs accordingly, and removes that fallback.
## Writer architecture and why iceberg-rust changes are not required
The relevant write hierarchy is:
Spark Iceberg V2 write
└── Spark-side IcebergWriteExec
└── IcebergWrite protobuf
└── Comet native IcebergWriteExec
└── iceberg-rust ParquetWriterBuilder
└── parquet-rs WriterProperties
More precisely:
1. Spark plans the Iceberg V2 write using Comet's split writer/committer
plan.
2. CometIcebergNativeWrite evaluates eligibility and serializes the write
into the IcebergWrite protobuf.
3. The Spark-side write is replaced with CometIcebergWriteExec.
4. The native planner converts the protobuf into the Rust IcebergWriteExec.
5. The Rust writer constructs parquet-rs WriterProperties.
6. Those properties are supplied to iceberg-rust's ParquetWriterBuilder.
The last step is visible in:
native/core/src/execution/operators/iceberg_write.rs:310
let parquet_builder =
ParquetWriterBuilder::new(writer_properties,
Arc::clone(&iceberg_schema));
This distinction is important. iceberg-rust has two relevant construction
paths:
ParquetWriterBuilder::new(writer_properties, iceberg_schema)
and:
ParquetWriterBuilder::from_table_properties(table_properties,
iceberg_schema)
At the iceberg-rust revision currently pinned by Comet,
from_table_properties only translates content-defined chunking properties. It
does not translate Iceberg
bloom-filter properties.
However, Comet does not call from_table_properties. It constructs
WriterProperties itself and passes them to ParquetWriterBuilder::new.
The pinned parquet-rs API already provides:
- set_column_bloom_filter_enabled
- set_column_bloom_filter_fpp
- set_column_bloom_filter_ndv
Therefore, per-column bloom-filter enablement does not require an
iceberg-rust contribution. Comet can configure the existing parquet-rs writer
directly.
This PR implements that path:
1. Read the resolved Iceberg table properties on the JVM.
2. Select write.parquet.bloom-filter-enabled.column.<column> entries whose
value is true, case-insensitively.
3. Strip the property prefix to obtain Iceberg dotted column paths.
4. Add those paths to IcebergParquetWriteSettings.
5. Convert the strings into parquet-rs ColumnPath values.
6. Call set_column_bloom_filter_enabled while constructing
WriterProperties.
7. Remove the corresponding native-write fallback rule.
Contributing broader property translation to iceberg-rust's
from_table_properties would still be useful for iceberg-rust consumers. It
would not automatically
change Comet, however, because Comet supplies its own WriterProperties.
## Scope and remaining limitations
This PR supports per-column enablement only:
write.parquet.bloom-filter-enabled.column.<column>
The other bloom-filter settings remain outside this PR's scope.
Feature Supported by
this PR Upstream work required?
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
━━━━━━━━━━━━━━━━━━━━━━━━
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Per-column bloom-filter enablement
Yes No
────────────────────────────────────────────────────────
────────────────────────
─────────────────────────────────────────────────────────────────────────────
Per-column FPP configuration
No No parquet-rs feature is needed; Comet still needs JVM/protobuf
translation
and parity validation
────────────────────────────────────────────────────────
────────────────────────
─────────────────────────────────────────────────────────────────────────────
Per-column NDV configuration
No No parquet-rs feature is needed; Comet still needs JVM/protobuf
translation
and parity validation
────────────────────────────────────────────────────────
────────────────────────
─────────────────────────────────────────────────────────────────────────────
Exact write.parquet.bloom-filter-max-bytes parity No; remains a
fallback A direct parquet-rs size-cap feature, or an equivalent upstream
mechanism,
is likely required
────────────────────────────────────────────────────────
────────────────────────
─────────────────────────────────────────────────────────────────────────────
Automatic Iceberg-property translation in iceberg-rust
No Optional and useful upstream, but not required by Comet's current
construction path
write.parquet.bloom-filter-max-bytes is materially different from the
enablement property.
Iceberg/parquet-mr applies it as a hard maximum size for generated bloom
filters. parquet-rs currently sizes bloom filters from the expected number of
distinct
values and the requested false-positive probability. Its WriterProperties
API does not expose an equivalent hard byte cap.
For that reason, this PR deliberately preserves the existing fallback when
write.parquet.bloom-filter-max-bytes is explicitly configured. Silently
accepting that
property would produce files with different sizing semantics from
iceberg-java.
The following settings also remain unsupported and continue to fall back:
- write.parquet.bloom-filter-fpp.column.<column>
- write.parquet.bloom-filter-ndv.column.<column>
- write.parquet.bloom-filter-adaptive-enabled
- write.parquet.bloom-filter-max-bytes
## What changes are included in this PR?
### JVM-to-native configuration propagation
IcebergWriteProtoTranslation now reads the Iceberg per-column bloom-filter
prefix using Iceberg's runtime table-property constant.
Properties are filtered so that:
- true enables the column, case-insensitively.
- false does not enable the column.
- Column names are sorted before protobuf serialization to make the
resulting message deterministic.
- Dotted names are preserved for nested Parquet column paths.
The enabled paths are added to a new repeated field in
IcebergParquetWriteSettings:
repeated string bloom_filter_enabled_columns = 8;
Relevant files:
-
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala
- native/proto/src/proto/operator.proto
### Native writer configuration
build_writer_properties now converts every enabled path into a parquet-rs
ColumnPath and configures the writer with:
builder =
builder.set_column_bloom_filter_enabled(column.as_str().into(), true);
Relevant file:
- native/core/src/execution/operators/iceberg_write.rs
### Eligibility behavior
The rule that rejected every enabled per-column bloom filter has been
removed.
A table with:
write.parquet.bloom-filter-enabled.column.id=true
is now considered compatible with the native Iceberg writer.
The allowlist still rejects write.parquet.bloom-filter-max-bytes and the
other unsupported bloom-filter properties described above.
Relevant file:
-
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
### Documentation
The native Iceberg write property matrix now records that per-column
bloom-filter enablement accepts both true and false, and that bloom filters are
written only
for columns set to true.
It continues to document write.parquet.bloom-filter-max-bytes as an
unvetted property that triggers fallback.
Relevant file:
- docs/source/user-guide/latest/iceberg-writes.md
## How are these changes tested?
### 1. JVM property-to-protobuf translation
Test:
IcebergWriteProtoTranslationSuite:
per-column bloom filter properties are translated deterministically
This unit test supplies:
- region=TRUE
- id=true
- amount=false
It asserts that the protobuf contains exactly:
[id, region]
This verifies:
- Case-insensitive handling of true.
- Exclusion of explicitly disabled columns.
- Correct removal of the Iceberg property prefix.
- Deterministic ordering in the protobuf.
- The complete JVM-to-protobuf configuration handoff.
Without this test, the write could be marked compatible while the
requested columns were silently lost before native execution.
### 2. Native WriterProperties construction
Test:
execution::operators::iceberg_write::tests::
enables_bloom_filters_only_for_configured_columns
This Rust unit test configures:
- Flat column id.
- Nested path nested.value.
- Unconfigured control column other.
It asserts that parquet-rs exposes bloom-filter properties for id and
nested.value, but not for other.
This verifies:
- Protobuf settings reach build_writer_properties.
- Flat Iceberg column names become valid parquet-rs paths.
- Dotted nested paths become valid parquet-rs ColumnPath values.
- Bloom filters are enabled selectively rather than globally.
- Unconfigured columns retain parquet-rs's default of no bloom filter.
Without this test, a translation error could enable bloom filters
globally, ignore nested paths, or fail to configure parquet-rs at all.
### 3. Native write eligibility
Tests:
CometIcebergWriteDetectionSuite:
Compatible when a per-column bloom filter is enabled
and:
CometIcebergWriteDetectionSuite:
fall-back: write.parquet.bloom-filter-max-bytes set
The first test verifies that enabling a bloom filter no longer rejects the
native writer during planning.
The second is an important scope guard: it verifies that this PR does not
accidentally accept write.parquet.bloom-filter-max-bytes, whose exact parquet-mr
semantics are not implemented.
Together, these tests distinguish the newly supported property from the
remaining unsupported byte-cap behavior.
### 4. End-to-end native Iceberg write and physical Parquet verification
Test:
CometIcebergWriteActionSuite:
native acceleration: writes configured Iceberg Parquet bloom filters
This is the end-to-end correctness test.
It creates an Iceberg table with:
write.parquet.bloom-filter-enabled.column.id=true
and inserts 256 rows while native Iceberg writes are enabled.
The test then verifies four independent invariants.
#### Native execution was actually selected
The executed Spark plans are captured and searched for
CometIcebergWriteExec.
The assertion requires at least one real executed CometIcebergWriteExec,
following the same general approach used by the native Iceberg scan tests.
Checking only
eligibility or an optimized plan would not prove that execution reached
the native writer after AQE and physical-plan conversion.
#### Exactly one Iceberg snapshot was committed
The snapshot count must advance by exactly one.
This ensures that plan capture or AQE replanning did not cause duplicate
commits and that the native write completed through the normal Iceberg commit
path.
#### Written data is correct
The resulting table is read back and all expected IDs are compared.
This guards against a writer that produces bloom metadata but corrupts,
drops, or duplicates actual table rows.
#### Bloom filters physically exist in every generated Parquet file
The test reads the Iceberg data_files metadata table, opens every current
Parquet data file with ParquetFileReader, and examines every column chunk.
For the configured id column it asserts:
bloomFilter != null
bloomFilter.getBitsetSize > 0
For every unconfigured column it asserts:
bloomFilter == null
This is the strongest bloom-filter correctness check in the PR. It proves
that:
- The configuration was not merely accepted by the eligibility gate.
- The configuration survived JVM-to-protobuf translation.
- The native writer applied it to parquet-rs.
- parquet-rs physically serialized a non-empty bloom filter.
- Bloom filters were not unintentionally enabled for other columns.
### 5. Existing native scan coverage
CometIcebergNativeSuite already contains:
filter on a table with a decimal column does not fail the native scan
That test creates an Iceberg table with bloom filters on several primitive
and decimal columns, writes rows, runs a wide equality predicate, checks answer
parity
with Spark, and asserts exactly one CometIcebergNativeScanExec.
That test covers reading and pruning a bloom-filter-bearing table through
the native Iceberg scan. It does not cover native bloom-filter writes: it
enables
native Iceberg scanning, while the data may still be produced by the JVM
Iceberg writer.
The new CometIcebergWriteActionSuite test complements it by proving that
Comet itself can produce the bloom filters and that CometIcebergWriteExec was
used.
## Verification performed
make core
Passed.
cargo test -p datafusion-comet \
execution::operators::iceberg_write::tests::enables_bloom_filters_only_for_configured_columns
Passed: 1 test.
./mvnw test -Dtest=none \
-Dsuites="org.apache.comet.serde.operator.IcebergWriteProtoTranslationSuite,org.apache.comet.CometIcebergWriteDetectionSuite"
Passed: 67 tests.
./mvnw test -Dtest=none \
-Dsuites="org.apache.comet.CometIcebergWriteActionSuite native
acceleration: writes configured Iceberg Parquet bloom filters"
Passed: 1 test.
The Maven tests were run with JDK 17, which is supported by the project.
The machine's default JDK 25 is incompatible with the repository's pinned
JaCoCo and
Google Java Format versions.
The following checks also passed:
cargo fmt --all -- --check
git diff --check
Spotless
Scalastyle
Issue references:
[#5643](https://github.com/apache/datafusion-comet/issues/5643),
[#5649](https://github.com/apache/datafusion-comet/issues/5649), [#5304]
(https://github.com/apache/datafusion-comet/issues/5304),
[#5298](https://github.com/apache/datafusion-comet/pull/5298), and
[#5361](https://github.com/apache/
datafusion-comet/pull/5361).
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases. You can
link an issue to this PR using the GitHub syntax. For example `Closes #123`
indicates that this PR will close issue #123.
-->
Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in
the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your
changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is
sometimes worth providing a summary of the individual changes in this PR.
-->
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are
they covered by existing tests)?
-->
Assisted by Codex.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
