You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/05 14:12:22 UTC
Re: [PR] feat: support bloom filters in native Iceberg writes [datafusion-comet]
andygrove commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4184946253
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -600,6 +604,494 @@ class CometIcebergWriteActionSuite
}
}
+ test("native acceleration: writes configured Iceberg Parquet bloom filters") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(
+ warehouseDir,
+ "native_bloom",
+ partitionSpec = "",
+ properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
+
+ val expectedIds = 0 until 256
+ assertNativeWriteEngages("native_bloom", expectedIds) {
+ spark.sql(
+ "INSERT INTO cat.db.native_bloom " +
+ "SELECT CAST(id AS INT), CONCAT('region-', CAST(id % 4 AS STRING)), " +
+ "CAST(id AS DOUBLE) FROM range(256)")
+ }
+
+ assertParquetBloomFilters("native_bloom", enabledColumns = Set("id"))
+ }
+ }
+
+ test("enabled=false plus NDV matches JVM behavior for the Iceberg runtime") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val configuredFpp = 0.01
+ val configuredNdv = 1000
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ s"'write.parquet.bloom-filter-fpp.column.id'='$configuredFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'")
+ createTable(warehouseDir, "bloom_false_ndv_native", partitionSpec = "", properties)
+ createTable(warehouseDir, "bloom_false_ndv_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) " +
+ s"FROM range(0, $configuredNdv, 1, 1)")
+
+ assertNativeWriteEngages("bloom_false_ndv_native", 0 until configuredNdv) {
+ insert("bloom_false_ndv_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert("bloom_false_ndv_jvm")
+ }
+
+ val nativeHasBloom = parquetBloomFilterPresent("bloom_false_ndv_native", "id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_false_ndv_jvm", "id")
+ assert(nativeHasBloom == jvmHasBloom)
+ if (icebergSupportsBloomNdv) {
+ assert(nativeHasBloom, "an interpreted NDV must re-enable the Bloom filter")
+ val native = parquetBloomFilterBytes("bloom_false_ndv_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_false_ndv_jvm", "id")
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ assert(!nativeHasBloom, "an uninterpreted NDV must not override enabled=false")
+ }
+ }
+ }
+
+ test("enabled=false preserves JVM validation errors for malformed FPP and NDV") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ Seq(
+ ("fpp", "'write.parquet.bloom-filter-fpp.column.id'='garbage'", icebergSupportsBloomFpp),
+ ("ndv", "'write.parquet.bloom-filter-ndv.column.id'='garbage'", icebergSupportsBloomNdv))
+ .foreach { case (suffix, malformedProperty, interpreted) =>
+ val properties =
+ Some(s"'write.parquet.bloom-filter-enabled.column.id'='false', $malformedProperty")
+ val nativeTable = s"bloom_false_bad_${suffix}_native"
+ val jvmTable = s"bloom_false_bad_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(warehouseDir, table, partitionSpec = "", properties = properties)
+ }
+
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table VALUES (1, 'region', 1.0)")
+
+ if (interpreted) {
+ val withComet = intercept[Throwable] {
+ withNativeEnabled(insert(nativeTable))
+ }
+ val withJvm = intercept[Throwable] {
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert(jvmTable)
+ }
+ }
+ val cometCause = exceptionChain(withComet).last
+ val jvmCause = exceptionChain(withJvm).last
+ assert(cometCause.getClass == jvmCause.getClass)
+ assert(cometCause.getMessage == jvmCause.getMessage)
+ } else {
+ assertNativeWriteEngages(nativeTable, Seq(1))(insert(nativeTable))
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ insert(jvmTable)
+ }
+ assert(
+ !parquetBloomFilterPresent(nativeTable, "id") &&
+ !parquetBloomFilterPresent(jvmTable, "id"),
+ s"uninterpreted $suffix must not enable a Bloom filter")
+ }
+ }
+ }
+ }
+
+ test("max-bytes=32 falls back only when parquet-mr ignores the cap") {
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ val minimumBytes = 32
+ val naturallyMinimumNdv = 1
+ val bindingNdv = 1000000
+ val bindingFpp = 0.0001
+ val parquetMrBindingBytes = 4 * 1024 * 1024
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true', "
+ createTable(
+ warehouseDir,
+ "bloom_minimum_native",
+ partitionSpec = "",
+ properties = Some(
+ enabled + s"'write.parquet.bloom-filter-ndv.column.id'='$naturallyMinimumNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ createTable(
+ warehouseDir,
+ "bloom_minimum_fallback",
+ partitionSpec = "",
+ properties = Some(
+ enabled + s"'write.parquet.bloom-filter-fpp.column.id'='$bindingFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$bindingNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+
+ assertNativeWriteEngages("bloom_minimum_native", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_native VALUES (1, 'region', 1.0)")
+ }
+ assertNativeWriteDoesNotEngage("bloom_minimum_fallback", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_fallback VALUES (1, 'region', 1.0)")
+ }
+
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_native", "id").forall(_.length == minimumBytes))
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_fallback", "id")
+ .forall(_.length == parquetMrBindingBytes))
+ }
+ }
+
+ test("native bloom filters resolve list and map leaves to physical Parquet paths") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.native_nested_bloom (
+ id INT,
+ tags ARRAY<STRING>,
+ attrs MAP<STRING, INT>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.element'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.key'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.value'='true'
+ )
+ """)
+
+ assertNativeWriteEngages("native_nested_bloom", Seq(1, 2)) {
+ spark.sql("""
+ INSERT INTO cat.db.native_nested_bloom VALUES
+ (1, array('red', 'green'), map('small', 10, 'large', 20)),
+ (2, array('blue'), map('medium', 30))
+ """)
+ }
+
+ assertParquetBloomFilters(
+ "native_nested_bloom",
+ enabledColumns = Set("tags.list.element", "attrs.key_value.key", "attrs.key_value.value"))
+ }
+ }
+
+ test("quoted bloom-filter columns renamed by Iceberg fall back to the JVM footer") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("bloom_quoted_name", "bloom_quoted_name_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ `order id` INT
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.order id'='true'
+ )
+ """)
+ }
+
+ val snapshot = withNativeEnabled {
+ captureWrite("bloom_quoted_name") {
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name VALUES (1), (2)")
+ }
+ }
+ assertExactlyOneCommit(snapshot)
+ val nativeExecs = snapshot.plans.flatMap { plan =>
+ collectWithSubqueries(plan) { case exec: CometIcebergWriteExec => exec }
+ }
+ assert(
+ nativeExecs.isEmpty,
+ s"expected the sanitized Bloom-filter path to fall back, plans:\n${snapshot.plans.mkString("\n--\n")}")
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") {
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name_jvm VALUES (1), (2)")
+ }
+ val fallbackHasBloom = parquetBloomFilterPresent("bloom_quoted_name", "order_x20id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_quoted_name_jvm", "order_x20id")
+ assert(
+ fallbackHasBloom == jvmHasBloom,
+ "expected the fallback write to match the Parquet Java footer")
+ if (icebergSupportsBloomFpp) {
+ assert(jvmHasBloom, "expected this Iceberg version to write the quoted-column filter")
+ }
+ }
+ }
+
+ test("native bloom filter is byte-identical to parquet-mr when max-bytes binds") {
Review Comment:
Every byte-identity and membership check in these tests uses the `id INT` column, so parity is only shown for INT32. A hash mismatch on any other physical type would be a false negative. Iceberg's Bloom reader would skip row groups that hold the value, and every later read of the table would silently lose rows. By source the two libraries hash the same raw bytes today, but nothing here pins that across a parquet-rs bump. Iceberg's own `TestSparkReaderWithBloomFilter` can't cover it either, because it writes through the Java `DataWriter` and never reaches the native writer.
Could we add a sibling of this test with one Bloom-enabled column per type the Iceberg reader prunes? That means bigint, float, double, string, binary, date, timestamp, and decimals stored as INT32, INT64 and fixed. With only `max-bytes=4096` set and 10,000 distinct values per column, both writers allocate the cap and the filters are far too dense for parquet-rs to fold, so the bytes should match column by column. Leaving out NDV and FPP also lets it run on Iceberg 1.5.2.
##########
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:
When the opt-in lands, could this section move out from under "Accepted divergences behind the toggle", or could that intro change? It says enabling the native writer accepts every difference listed and that they fall into three classes. Folding would then sit behind its own key, and it changes pruning, so it isn't cosmetic. The opt-in's doc should also cover boolean columns. parquet-mr never inserts boolean values (`ColumnValueCollector.write(boolean, ...)` only updates statistics), so Java writes an all-zero filter where parquet-rs writes a populated one.
##########
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:
+1. I checked this against the four runtime jars. Each one parses `write.parquet.bloom-filter-max-bytes` in `Parquet.WriteBuilder.Context` and fails its `> 0` check on every Parquet write, before it looks at any Bloom column. So a table with `0` or `-1` and no Bloom column fails on the JVM path today and would write natively with this change. Could the `configured.isEmpty` branch reject parsed values `<= 0` too?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org