peterxcli commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3999193648
##########
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala:
##########
@@ -112,6 +162,28 @@ object IcebergWriteProtoTranslation {
parseJavaInt(props, Keys.ParquetDictSizeBytes,
Defaults.DictSizeBytes.toInt).toLong
val pageRowLimit = parseJavaInt(props, Keys.ParquetPageRowLimit,
Defaults.PageRowLimit)
val compression = resolveCompression(props)
+ // Iceberg properties use logical schema paths, while Parquet writer
properties require the
+ // physical leaf path. Missing fields are skipped, matching Iceberg Java's
writer behavior.
+ val bloomFilterColumns = enabledBloomFilterColumnNames(props)
+ .flatMap { icebergName =>
+ parquetPathByIcebergColumnName.get(icebergName).map(icebergName -> _)
Review Comment:
Could we preserve Iceberg 1.5.2's physical-path property names here, or fall
back when they are enabled? For `tags ARRAY<STRUCT<a: INT>>` and `attrs
MAP<STRING, STRUCT<b: INT>>`, its Java writer creates Bloom filters for these
settings:
```properties
write.parquet.bloom-filter-enabled.column.tags.list.element.a=true
write.parquet.bloom-filter-enabled.column.attrs.key_value.value.b=true
```
I checked the Java file footers and ran this PR's Scala path resolver
against the 1.5.2 runtime: neither name exists in the canonical-name map, so
this lookup silently drops both filters. `schema.findField` also returns null
for these physical paths, so that check alone would miss this case.
Please cover both leaves with a native/JVM footer comparison on Iceberg
1.5.2, alongside the existing 1.8.1 alias issue.
##########
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:
The proposed opt-in for native Bloom writes sounds reasonable. Could this
paragraph explicitly say that folding trades smaller files for potentially
higher realised FPP and less pruning than Java? A table-level NDV estimate can
exceed a row group's actual NDV because of skew or a small final row group, so
this does not require a badly configured table.
When documenting the new option, please also make clear that disabling it
selects the JVM writer for Bloom-enabled tables; it cannot disable folding
within parquet-rs yet.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -600,6 +604,453 @@ 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") {
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // This NDV/FPP pair requests a 128 MiB allocation before max-bytes is
applied. The 4 KiB
+ // maximum must therefore bind on both writers rather than merely
coinciding with the
+ // naturally selected size.
+ val configuredNdv = 100000000L
+ val bindingMaxBytes = 4 * 1024
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='0.01', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$bindingMaxBytes'")
+ createTable(warehouseDir, "bloom_identity_native", partitionSpec = "",
properties)
+ createTable(warehouseDir, "bloom_identity_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) FROM range(0,
10000, 1, 1)")
+
+ assertNativeWriteEngages("bloom_identity_native", 0 until 10000) {
+ insert("bloom_identity_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("bloom_identity_jvm")
+ }
+
+ val native = parquetBloomFilterBytes("bloom_identity_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_identity_jvm", "id")
+ assert(
+ native.nonEmpty && native.forall(_.length == bindingMaxBytes),
+ s"native Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(
+ jvm.nonEmpty && jvm.forall(_.length == bindingMaxBytes),
+ s"JVM Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(native.size == jvm.size, "native and JVM writes must produce the
same file count")
+ assert(
+ native.zip(jvm).forall { case (left, right) =>
java.util.Arrays.equals(left, right) },
+ "expected byte-identical capped SBBF bitsets for identical values and
allocation")
+ }
+ }
+
+ test("native bloom sizing matches JVM for shape properties supported by the
Iceberg runtime") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true'"
+ val cases: Seq[(String, String, Boolean, Boolean)] = Seq(
+ (
+ "fpp_only",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.02'",
+ icebergSupportsBloomFpp,
+ false),
+ (
+ "ndv_only",
+ s"$enabled, 'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomNdv,
+ true),
+ (
+ "both",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.005', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true),
+ // The requested NDV/FPP needs far more than 64 bytes. Like
parquet-mr, max wins and the
+ // target FPP becomes impossible to guarantee, but membership must
remain correct.
+ (
+ "binding_cap",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.0001', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000000', " +
+ "'write.parquet.bloom-filter-max-bytes'='64'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true))
+
+ cases.filter(_._3).foreach { case (suffix, properties, _,
expectByteIdentity) =>
+ val nativeTable = s"bloom_shape_${suffix}_native"
+ val jvmTable = s"bloom_shape_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(warehouseDir, table, partitionSpec = "", properties =
Some(properties))
+ }
+ // Keep the explicit-NDV cases at their estimated cardinality so
parquet-rs does not
+ // fold their allocation before byte-parity is checked.
+ val insertedCardinality = if (suffix == "ndv_only" || suffix ==
"both") 1000 else 256
+ val ids = 0 until insertedCardinality
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table SELECT CAST(id AS INT),
'region', " +
+ s"CAST(id AS DOUBLE) FROM range(${ids.start}, ${ids.end}, 1, 1)")
+
+ assertNativeWriteEngages(nativeTable, ids) {
+ insert(nativeTable)
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key ->
"false") {
+ insert(jvmTable)
+ }
+ val native = parquetBloomFilterBytes(nativeTable, "id")
+ val jvm = parquetBloomFilterBytes(jvmTable, "id")
+ if (expectByteIdentity) {
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ // Without explicit NDV, parquet-rs can fold to the observed
cardinality while Parquet
Review Comment:
A concrete regression case for this: Iceberg 1.11, `ndv=1000`, `fpp=0.01`,
inserting INT32 values `0..99`. In a component probe using a real Java-written
filter and the extracted parquet-rs 59.3.0 folding methods, Java kept 2,048
bytes and folding produced 128 bytes. Probing the 1,000,000 absent values
`10_000_000..10_999_999` produced 0 versus 9,088 false positives (0.9088%); all
inserted values remained present.
Could we add this fixture to the native/JVM write suite, assert native
execution, and compare serialized sizes and membership on both filters? This
would cover overestimated explicit NDV and document the pruning tradeoff. The
measurements above are from a component probe, not a full native writer run.
--
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]
