comphead commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4417471587
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -2310,8 +2313,147 @@ fn build_writer_properties(settings:
&IcebergParquetWriteSettings) -> DFResult<W
.set_dictionary_page_size_limit(settings.dict_size_bytes as usize)
.set_data_page_row_count_limit(settings.page_row_limit as usize)
.set_statistics_enabled(EnabledStatistics::Page)
- .set_statistics_truncate_length(None)
- .build())
+ .set_statistics_truncate_length(None);
+ for bloom_filter in &settings.bloom_filters {
+ let path = parquet_column_path(&bloom_filter.parquet_path);
+ let fpp = bloom_filter.fpp;
+ let ndv = bloom_filter.ndv;
+ let max_bytes = settings.bloom_filter_max_bytes as usize;
+ validate_bloom_filter_inputs(fpp, max_bytes)?;
+ let target_bytes = parquet_mr_bloom_filter_bytes(ndv, fpp, max_bytes);
+ let synthetic_ndv = synthetic_ndv_for_bloom_filter_bytes(target_bytes,
fpp)?;
+ 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);
Review 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 gu
ide's memory paragraph would change with it. I derived this from the source
and have not measured it.
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -653,13 +655,77 @@ unchanged.
### Accepted divergences
-Some differences between parquet-mr and the pinned parquet-rs / iceberg-rust
are unconditional —
-they apply to every native write and cannot be configured away. Enabling
-`spark.comet.write.iceberg.enabled` accepts them. They fall into two classes
with very
-different blast radius: differences confined to the physical bytes of a data
file (cosmetic —
-no reader decision is based on them), and differences visible in manifest
metadata (these
-outlive the write and feed later readers' pruning decisions, so each one is
analyzed
-individually below).
+Enabling `spark.comet.write.iceberg.enabled` accepts the physical file-layout
and manifest-metadata
+differences described below. Bloom-filter differences require the additional
+`spark.comet.write.iceberg.bloomFilterFolding.enabled` opt-in, because they
can affect pruning.
+
+#### Bloom filters
+
+Comet's native Iceberg writer supports tables with column Bloom filters.
However, byte-identical
+behavior to classic Spark cannot be guaranteed for all settings: parquet-rs
enables folding by
+default, while Spark's non-adaptive parquet-java writer does not.
+
+Folding automatically shrinks a filter when its initial allocation from NDV or
`max-bytes` is
+larger than the actual values need at the requested false-positive probability
(FPP). It saves
+serialized space by shrinking toward the target FPP; Java retains the larger
filter. Folding
+retains the original buffer capacity, so this does not imply lower peak writer
memory.
+
+In addition to the [general native-write eligibility
conditions](#native-parquet-write-eligibility),
+native writes for configured Bloom columns require all of the following:
+
+- `spark.comet.write.iceberg.enabled=true`.
+- `spark.comet.write.iceberg.bloomFilterFolding.enabled=true`.
+- The Iceberg runtime reports an identifiable version of 1.10 or later.
Earlier or
+ unidentified runtimes use Java whenever a per-column Bloom property is
configured,
+ including `false` enablement and standalone FPP/NDV properties. The opt-in
cannot enable native Bloom writes there.
+- The effective `write.parquet.bloom-filter-max-bytes` (default 1 MiB) is a
power of two from
+ 32 bytes through 128 MiB.
+- An interpreted NDV is absent or a positive Java long no greater than
`Long.MAX_VALUE / 8`.
+- Each interpreted FPP is finite, strictly between 0 and 1, and representable
at the required
+ native filter sizes.
+- With a 32-byte cap and interpreted explicit NDV, the NDV/FPP request fits in
32 bytes, because
+ Java ignores that boundary as a maximum for larger requests.
+- Bloom-filter columns use physical Parquet paths supported by the native
writer.
+- Bloom-filter columns are not FLOAT or DOUBLE, including nested leaves.
Floating-point columns
+ without Bloom filters remain eligible for native writes.
+
+The default Spark profiles have the following Bloom-write eligibility. The
loaded Iceberg
+version determines eligibility; Spark's version alone does not enable this
feature.
+
+| Spark profile | Pinned Iceberg | Bloom opt-in behavior
|
+| ------------- | -------------- |
--------------------------------------------- |
+| 3.4 | 1.5.2 | Java writer, even with the opt-in enabled
|
+| 3.5 | 1.8.1 | Java writer, even with the opt-in enabled
|
+| 4.0 | 1.10.0 | Native eligible; FPP interpreted, NDV
ignored |
+| 4.1 | 1.11.0 | Native eligible; FPP and NDV interpreted
|
+
+The cap range applies whenever a `write.parquet.bloom-filter-enabled.column.*`
property is
+present, even if set to `false`. With no such properties, the cap is unused:
unset or any
+positive Java int, such as `100`, keeps an otherwise eligible write native.
Malformed, zero,
+or negative caps select Java, which rejects them through its normal validation.
+
+The Bloom option defaults to `false`, selecting Java's writer for
Bloom-enabled tables. Enabling
+it accepts byte differences and the pruning tradeoff below; it cannot turn
folding off inside
+the pinned parquet-rs writer.
+Enabling it is generally safe when the requested FPP meets your pruning needs:
folding preserves
+inserted-value membership and introduces no false negatives. Unsupported
settings select Java's
+writer, where normal property validation applies. FLOAT/DOUBLE Bloom writes
use Java until native
+writes handle NaN equality consistently with Spark. The fallback is decided
from the schema and
+effective write properties before execution, and its reason identifies the
affected columns.
+
+Pruning can regress when existing Java-written filters are oversized for their
actual values
+and therefore have a much lower FPP than requested. For example, with no
explicit NDV, a 64-MiB
+maximum and only 256 distinct values, Java can retain a 64-MiB filter with
extra pruning accuracy.
+Rust can shrink that filter toward the target FPP and produce more false
positives. Leave the
+option off if your workload relies on that extra accuracy. This can happen
with an overestimated
+NDV, skew, or a partly filled row group. With explicit NDV, `max-bytes` is a
ceiling; a large cap
+alone does not enlarge the allocation.
+
+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
Review 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.
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -39,6 +39,13 @@ import org.apache.comet.util.ClassLoaders
*/
object IcebergReflection extends Logging {
+ case class ParquetPathResolution(
+ pathByIcebergColumnName: Map[String, String],
+ pathByFieldId: Map[Int, String],
+ physicalPaths: Set[String],
Review 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?
--
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]
