comphead commented on code in PR #7226:
URL: https://github.com/apache/datafusion-comet/pull/7226#discussion_r4417472005
##########
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:
This is the same enumeration as the inline `configured` in
`requireNativeSupportedBloomFilterProperties` and the private
`IcebergWriteProtoTranslation.configuredBloomFilterColumnNames`, apart from
`.sorted` in the last. Could the translation's version become
`private[operator]` and replace both?
##########
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:
This rule always returns `None` and declines only by letting
`bloomFilterParquetPaths` throw from `require`. `getSupportLevel`'s `NonFatal`
handler then reports "Iceberg native write detection failed: requirement
failed: Multiple Bloom column names resolve to the same Parquet leaf: ...", so
an intended decline reads like a reflection failure. Could
`bloomFilterParquetPaths` return `Either[String, Map[String, String]]`, so the
rule returns the collision or empty-name reason itself?
`requireNoFloatingPointBloomFilters` resolves the schema and paths again, and
`convert` a third time, so folding the FLOAT/DOUBLE check into this rule would
resolve once in the gate.
--
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]
