NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4034849778
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -283,6 +284,156 @@ 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 val MinBloomFilterBytes = 32
+ private 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 58.x represents Bloom filters as a power-of-two number of
bytes. 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 maxRejection =
Review Comment:
Thanks for spotting this! Fixed in commit `fix: ignore unused bloom max byte
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]
