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: [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]
