NikitaMatskevich commented on code in PR #5724: URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4372214610
########## docs/source/user-guide/latest/iceberg-writes.md: ########## @@ -231,6 +233,42 @@ no reader decision is based on them), differences visible in manifest metadata ( the write and feed later readers' pruning decisions, so each one is analyzed individually below), and one operational path-layout caveat. +### Parquet Bloom-filter sizing + +[Iceberg's documented write properties](https://iceberg.apache.org/docs/latest/configuration/#write-properties) +describe three related inputs. FPP is the requested false-positive probability (default `0.01`), +NDV is the expected number of distinct values when explicitly set, and `max-bytes` is an upper +bound (default 1 MiB). + +Apache Parquet Java permits arbitrary integer caps. When such a cap binds, it serializes exactly +that many bytes, although only complete 32-byte SBBF blocks are used and any trailing partial +block remains zero. The Apache Arrow Rust `parquet` crate requires a power-of-two block count so +its post-write folding remains valid. A non-power-of-two cap can therefore change the +hash-to-block mapping, making a filter that may have worse reader pruning than Parquet Java's +filter. + +Comet uses its native Iceberg writer only when the effective `max-bytes` value is a power of two +from 32 bytes through 128 MiB inclusive. If an explicit value is not a power of two or is outside +that range, `CometIcebergWriteExec` is not used for the write; Spark's default Iceberg Java writer +writes the table instead. The same fallback applies when `max-bytes=32` would bind an explicit +NDV/FPP request, because Parquet Java ignores exactly 32 bytes as a maximum, and when NDV is above +`Long.MAX_VALUE / 8`, where Parquet Java's sizing multiplication can overflow. + +For supported values, Apache Parquet Java applies the sizing properties as follows: + +- with no NDV, allocate the full `max-bytes` value; +- with an NDV, calculate a requested size from NDV and FPP, then cap it at `max-bytes`; +- when the cap binds, it takes precedence, so the requested FPP is not guaranteed; +- a large maximum never enlarges the allocation selected by an explicit, smaller NDV. + +For every write that is eligible for the native path, Comet applies exactly the same allocation +decision algorithm. + +After values are inserted, the Apache Arrow Rust `parquet` crate may fold a sparsely populated +filter to a smaller power-of-two filter while preserving the requested FPP. Parquet Java's Review Comment: Thank you, [this clarification](https://github.com/apache/datafusion-comet/pull/5724#discussion_r4184947730) is addressed in commit `docs: clarify Iceberg Bloom opt-in and unused caps`. The Accepted divergences intro names the separate Bloom opt-in; the Bloom section and config description explicitly cover Java's all-zero versus Rust's populated boolean filters. ########## spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala: ########## @@ -283,6 +284,180 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { private val requireNativeSupportedCompressionLevel: TriggerRule = ctx => IcebergWriteProtoTranslation.compressionLevelRejection(ctx.properties) + // These are Apache Parquet Java BlockSplitBloomFilter implementation bounds, not Iceberg + // TableProperties constants, so they cannot be obtained through IcebergReflection: + // scalastyle:off line.size.limit + // https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L40-L50 + // scalastyle:on line.size.limit + private[operator] val MinBloomFilterBytes = 32 + private[operator] val MaxBloomFilterBytes = 128 * 1024 * 1024 + private val BloomFilterHashProbes = 8 + private val MaxNonOverflowingBloomFilterNdv = Long.MaxValue / BloomFilterHashProbes + + /** + * Keep only Bloom shape properties interpreted by the Iceberg runtime on the classpath. Older + * Iceberg releases leave these table properties untouched but do not pass them to parquet-mr. + * Ignoring them here preserves that version's JVM-writer behavior while allowing the remaining + * supported Bloom configuration to execute natively. + */ + private def interpretedBloomFilterProperties( + properties: Map[String, String]): Map[String, String] = { + val unsupportedPrefixes = Seq( + PropertyKeys.ParquetBloomFilterColumnFppPrefix -> + "PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX", + PropertyKeys.ParquetBloomFilterColumnNdvPrefix -> + "PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX").collect { + case (prefix, constant) if IcebergReflection.tablePropertyConstantOpt(constant).isEmpty => + prefix + } + properties.filterNot { case (key, _) => unsupportedPrefixes.exists(key.startsWith) } + } + + /** + * parquet-rs 59.3.0 initially represents Bloom filters as a power-of-two number of bytes and + * may fold that allocation after values are inserted. parquet-mr accepts arbitrary caps and, + * when one binds, serializes that exact length. Keep those writes on the classic path instead + * of silently changing the number of usable Bloom blocks. + */ + private val requireNativeSupportedBloomFilterProperties: TriggerRule = ctx => { + val properties = interpretedBloomFilterProperties(ctx.properties) + val configuredMaxBytes = properties.get(PropertyKeys.ParquetBloomFilterMaxBytes).map { raw => + raw -> scala.util.Try(java.lang.Integer.parseInt(raw)).toOption + } + val malformedMaxRejection = configuredMaxBytes.collect { case (raw, None) => + s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw is not a Java int" + } + // Iceberg visits every enabled-prefix entry and applies enabled, FPP, then NDV. With no such + // entry, a parseable maximum cannot affect the file and must not force an otherwise-compatible + // write onto the classic path. Keep rejecting malformed values because protobuf translation + // parses this property before native execution. + val configured = properties.iterator.collect { + case (key, _) if key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) => + key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length) + }.toSeq + if (configured.isEmpty) { + malformedMaxRejection Review Comment: Thank you, [this documentation request](https://github.com/apache/datafusion-comet/pull/5724#issuecomment-5876500709) is addressed in commit `docs: clarify Iceberg Bloom opt-in and unused caps`. The eligibility table, Bloom section, and protobuf comment distinguish configured keys (including false) from unused positive caps such as 100, and retain Java validation for invalid caps. -- 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]
