NikitaMatskevich commented on code in PR #7226:
URL: https://github.com/apache/datafusion-comet/pull/7226#discussion_r4524032056
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -485,9 +485,11 @@ A write is eligible only when ALL of the following hold:
| `write.parquet.page-version`
| unset or `v1`
|
| `write.parquet.shred-variants`
| unset or
`false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write)
|
| `write.parquet.variant-inference-buffer-size`
| any value
(only meaningful when shredding, which is gated)
|
-| `write.parquet.bloom-filter-enabled.column.<col>`
| unset or
`false`
|
| `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`.
|
Review Comment:
Thank you, fixed in commit `docs: clarify Iceberg 1.6 and 1.7 Bloom
properties`. Also marked this PR as draft and added the 1.7.2 command in commit
`docs: document legacy Iceberg validation commands`.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -594,6 +605,329 @@ 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 => {
+ val prefixes = Seq(
+ PropertyKeys.BloomFilterColumnEnabledPrefix,
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix,
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix)
+ if (ctx.properties.keys.exists(key => prefixes.exists(key.startsWith))) {
+ bloomFilterRuntimeRejection(IcebergReflection.icebergVersion(),
ctx.properties)
+ } else {
+ None
+ }
+ }
+
+ private def bloomFilterVersion(icebergVersion: String): Option[(Int, Int)] =
{
+ if
(Option(icebergVersion).exists(_.matches("\\d+\\.\\d+\\.\\d+(?:[-+].+)?"))) {
+ val parts = icebergVersion.split("[.-]", 3)
+ scala.util.Try((parts(0).toInt, parts(1).toInt)).toOption
+ } else {
+ None
+ }
+ }
+
+ private[operator] def bloomFilterRuntimeRejection(
+ icebergVersion: String,
+ properties: Map[String, String] = Map.empty): Option[String] = {
+ bloomFilterVersion(icebergVersion) match {
+ case Some((1, minor))
+ if (minor == 6 || minor == 7) && properties.keys.exists(
+ _.startsWith(PropertyKeys.ParquetBloomFilterColumnFppPrefix)) =>
+ Some(
+ s"Iceberg $icebergVersion applies
${PropertyKeys.ParquetBloomFilterColumnFppPrefix}* " +
+ "independently; these settings require Java validation")
+ case Some((1, minor)) if minor >= 5 => None
+ case Some((1, _)) =>
+ Some(s"Native Bloom writes require Iceberg 1.5 or later; found
'$icebergVersion'")
+ case _ =>
+ Some(
+ s"Cannot determine Iceberg Bloom column-name convention for version
'$icebergVersion'")
+ }
+ }
+
+ private def configuredBloomColumns(properties: Map[String, String]):
Seq[String] =
Review Comment:
Thank you, addressed in commit `refactor: share configured Bloom column
discovery`.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -594,6 +605,329 @@ 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 => {
+ val prefixes = Seq(
+ PropertyKeys.BloomFilterColumnEnabledPrefix,
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix,
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix)
+ if (ctx.properties.keys.exists(key => prefixes.exists(key.startsWith))) {
+ bloomFilterRuntimeRejection(IcebergReflection.icebergVersion(),
ctx.properties)
+ } else {
+ None
+ }
+ }
+
+ private def bloomFilterVersion(icebergVersion: String): Option[(Int, Int)] =
{
+ if
(Option(icebergVersion).exists(_.matches("\\d+\\.\\d+\\.\\d+(?:[-+].+)?"))) {
+ val parts = icebergVersion.split("[.-]", 3)
+ scala.util.Try((parts(0).toInt, parts(1).toInt)).toOption
+ } else {
+ None
+ }
+ }
+
+ private[operator] def bloomFilterRuntimeRejection(
+ icebergVersion: String,
+ properties: Map[String, String] = Map.empty): Option[String] = {
+ bloomFilterVersion(icebergVersion) match {
+ case Some((1, minor))
+ if (minor == 6 || minor == 7) && properties.keys.exists(
+ _.startsWith(PropertyKeys.ParquetBloomFilterColumnFppPrefix)) =>
+ Some(
+ s"Iceberg $icebergVersion applies
${PropertyKeys.ParquetBloomFilterColumnFppPrefix}* " +
+ "independently; these settings require Java validation")
+ case Some((1, minor)) if minor >= 5 => None
+ case Some((1, _)) =>
+ Some(s"Native Bloom writes require Iceberg 1.5 or later; found
'$icebergVersion'")
+ case _ =>
+ Some(
+ s"Cannot determine Iceberg Bloom column-name convention for version
'$icebergVersion'")
+ }
+ }
+
+ private def configuredBloomColumns(properties: Map[String, String]):
Seq[String] =
+ properties.iterator.collect {
+ case (key, _) if
key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) =>
+ key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length)
+ }.toSeq
+
+ private val requireSupportedBloomFilterPaths: TriggerRule = ctx => {
+ // Resolve disabled entries too: a second spelling can disable or change
the same leaf.
+ val columns = configuredBloomColumns(ctx.properties)
+ if (columns.nonEmpty) {
+ val schema = IcebergReflection
+ .getWriteSchemaFromSparkWrite(ctx.sparkWrite)
+ .getOrElse(throw new IllegalStateException("Could not resolve Bloom
write schema"))
+ val resolution = IcebergReflection
+ .getParquetPathResolution(schema)
+ .getOrElse(throw new IllegalStateException("Could not resolve Bloom
Parquet paths"))
+ val _ = bloomFilterParquetPaths(columns, resolution, schema = schema)
Review Comment:
Thank you, addressed in commit `refactor: report Bloom path fallback reasons
directly`.
--
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]
