andygrove commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4388829361
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -594,6 +604,268 @@ 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"
+ case (raw, Some(value)) if value <= 0 =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw must be positive"
+ }
+ // Iceberg visits every enabled-prefix entry and applies enabled, FPP,
then NDV. With no such
+ // entry, a positive, parseable maximum cannot affect the file and must
not force an
+ // otherwise-compatible write onto the classic path. Keep rejecting
malformed and
+ // non-positive values because Iceberg validates them even with no Bloom
column.
+ val configured = properties.iterator.collect {
+ case (key, _) if
key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) =>
+ key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length)
+ }.toSeq
+ if (configured.isEmpty) {
+ malformedMaxRejection
+ } else {
+ val maxRejection = malformedMaxRejection.orElse {
+ configuredMaxBytes.collect {
+ case (_, Some(value))
+ if value < MinBloomFilterBytes || value > MaxBloomFilterBytes ||
+ (value & (value - 1)) != 0 =>
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$value must be a
power of two " +
+ s"in [$MinBloomFilterBytes, $MaxBloomFilterBytes] for native
writes"
+ }
+ }
+
+ maxRejection.orElse {
+ val maxBytes = configuredMaxBytes
+ .flatMap(_._2)
+ .getOrElse(IcebergWriteProtoTranslation.Defaults.BloomFilterMaxBytes)
+ // Validate associated shape properties even for enabled=false; a
valid NDV re-enables
+ // the filter in parquet-mr.
+ configured.iterator
+ .flatMap { column =>
+ val fppKey = PropertyKeys.ParquetBloomFilterColumnFppPrefix +
column
+ val ndvKey = PropertyKeys.ParquetBloomFilterColumnNdvPrefix +
column
+ val parsedFpp = properties.get(fppKey) match {
+ case Some(raw) =>
scala.util.Try(java.lang.Double.parseDouble(raw)).toOption
+ case None =>
Some(IcebergWriteProtoTranslation.Defaults.BloomFilterFpp)
+ }
+ val parsedNdv = properties
+ .get(ndvKey)
+ .flatMap(raw =>
scala.util.Try(java.lang.Long.parseLong(raw)).toOption)
+ val fppError = properties.get(fppKey).flatMap { raw =>
+ parsedFpp match {
+ case Some(value)
+ if value > 0.0d && value < 1.0d &&
java.lang.Double.isFinite(value) &&
+ bloomFilterSizesRepresentable(maxBytes, value) =>
+ None
+ case Some(value)
+ if value > 0.0d && value < 1.0d &&
java.lang.Double.isFinite(value) =>
+ Some(s"$fppKey=$raw cannot represent the configured native
Bloom sizes")
+ case _ => Some(s"$fppKey=$raw must be a finite double strictly
between 0 and 1")
+ }
+ }
+ val ndvError = properties.get(ndvKey).flatMap { raw =>
+ parsedNdv match {
+ case Some(value) if value > 0L && value <=
MaxNonOverflowingBloomFilterNdv => None
+ case Some(value) if value > MaxNonOverflowingBloomFilterNdv =>
+ Some(s"$ndvKey=$raw exceeds
$MaxNonOverflowingBloomFilterNdv; " +
+ "parquet-mr Bloom sizing may overflow")
+ case _ => Some(s"$ndvKey=$raw must be a positive Java long")
+ }
+ }
+ val ignoredMinimumCapError = parsedNdv.collect {
+ case ndv
+ if maxBytes == MinBloomFilterBytes && ndv > 0L &&
+ ndv <= MaxNonOverflowingBloomFilterNdv &&
+ parsedFpp.exists(
+ parquetMrRequestedBloomFilterBytes(ndv, _) >
MinBloomFilterBytes) =>
+
s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$MinBloomFilterBytes is ignored by
" +
+ s"parquet-mr for $ndvKey=$ndv"
+ }
+ Seq(fppError, ndvError, ignoredMinimumCapError).flatten
+ }
+ .toSeq
+ .headOption
+ }
+ }
+ }
+
+ private val requireSupportedBloomFilterRuntime: TriggerRule = ctx => {
Review Comment:
On Iceberg 1.5 and 1.8 this rule rejects as soon as any key with a Bloom
prefix is present, including
`write.parquet.bloom-filter-enabled.column.id=false`. In `branch-1.2` that
property is vetted and the write stays native, so a table that only sets
`enabled=false` on Spark 3.4 or 3.5 goes from native to Java with this change.
`legacy runtime rejects every configured Bloom entry before filtering disabled
names` and `legacy Bloom settings use Java despite the native folding opt-in`
pin that behavior. A property that writes no filter has nothing for the runtime
check to protect, and the empty-suffix and unknown-column keys are skipped by
every runtime anyway.
Could the runtime check apply only when
`hasEnabledBloomFilters(interpretedBloomFilterProperties(ctx.properties))` is
true? Then `enabled=false` and unmatched entries keep their current native
behavior on older runtimes, and standalone FPP and NDV keys still fall back
through the unvetted-key rule there. If you would rather keep this PR narrow
and let #7226 restore it, the two PRs need to land together so no release ships
the regression.
--
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]
