NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4372167769
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -890,72 +1012,15 @@ class CometIcebergWriteDetectionSuite extends
CometTestBase with CometIcebergTes
}
}
- /**
- * Runs Spark's transition insertion followed by
[[EliminateRedundantTransitions]] over a
- * hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source`
plan and returns the
- * write's final child.
- *
- * Hand-built rather than driven through SQL because the shape depends on
- * `spark.comet.sparkToColumnar.enabled` admitting the write's source
operator, and the set of
- * admitted operators is itself configurable. What matters is the rule's
behaviour at that
- * boundary, which this pins directly.
- */
- private def writeChildAfterTransitionRules(source: SparkPlan): SparkPlan = {
- val write = CometIcebergWriteExec(
- Operator.newBuilder().build(),
- CometSparkToColumnarExec(source),
- batchWrite = null,
- table = null,
- partitionSpecId = 0)
- val withTransitions = ApplyColumnarRulesAndInsertTransitions(Seq.empty,
false).apply(write)
- // Spark must insert a columnar-to-row transition below the row-based
write; if it stops doing
- // so the rest of the assertion is vacuous.
- assert(
-
withTransitions.asInstanceOf[CometIcebergWriteExec].child.isInstanceOf[ColumnarToRowExec],
- s"expected an inserted ColumnarToRowExec below the write,
got:\n$withTransitions")
- EliminateRedundantTransitions(spark)
- .apply(withTransitions)
- .asInstanceOf[CometIcebergWriteExec]
- .child
- }
-
- // https://github.com/apache/datafusion-comet/issues/5689: the write's input
transition has to
- // be stripped before the generic
`ColumnarToRowExec(CometSparkToColumnarExec)` cancellation
- // consumes it, otherwise that arm removes the Arrow bridge the write's FFI
input depends on.
- // Both source representations are covered because the cancellation treats
them differently:
- // over a row source it drops the bridge outright, over a Spark-columnar
source it keeps a
- // transition but leaves the write reading Spark `ColumnarVector`s instead
of `CometVector`s.
- test("row source keeps its Arrow bridge under the native Iceberg write") {
- val source = TransitionProbeLeaf(columnar = false)
- val child = writeChildAfterTransitionRules(source)
- assert(
- child == CometSparkToColumnarExec(source),
- s"expected the write to sit directly on CometSparkToColumnarExec,
got:\n$child")
- }
-
- test("Spark-columnar source keeps its Arrow bridge under the native Iceberg
write") {
- val source = TransitionProbeLeaf(columnar = true)
- val child = writeChildAfterTransitionRules(source)
- assert(
- child == CometSparkToColumnarExec(source),
- s"expected the write to sit directly on CometSparkToColumnarExec,
got:\n$child")
+ private def assumeIcebergBloomShapeProperties(): Unit = {
+ assume(
+ IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX")
Review Comment:
Thank you, this was fixed in commit `test: gate Iceberg bloom shape
properties independently`.
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -665,8 +666,146 @@ fn build_writer_properties(settings:
&IcebergParquetWriteSettings) -> DFResult<W
.set_dictionary_page_size_limit(settings.dict_size_bytes as usize)
.set_data_page_row_count_limit(settings.page_row_limit as usize)
.set_statistics_enabled(EnabledStatistics::Page)
- .set_statistics_truncate_length(None)
- .build())
+ .set_statistics_truncate_length(None);
+ for column in &settings.bloom_filter_enabled_columns {
+ let path = parquet_column_path(column);
+ let fpp = settings
+ .bloom_filter_fpp_by_column
+ .get(column)
+ .copied()
+ .unwrap_or(ICEBERG_DEFAULT_BLOOM_FILTER_FPP);
+ let ndv = settings.bloom_filter_ndv_by_column.get(column).copied();
+ let max_bytes = settings.bloom_filter_max_bytes as usize;
+ validate_bloom_filter_inputs(fpp, max_bytes)?;
+ let target_bytes = parquet_mr_bloom_filter_bytes(ndv, fpp, max_bytes);
+ let synthetic_ndv = synthetic_ndv_for_bloom_filter_bytes(target_bytes,
fpp)?;
+ builder = builder
+ .set_column_bloom_filter_enabled(path.clone(), true)
+ .set_column_bloom_filter_fpp(path.clone(), fpp)
+ .set_column_bloom_filter_max_ndv(path, synthetic_ndv);
Review Comment:
Thank you, this was addressed in commit `feat: make native Iceberg bloom
folding opt-in`. Upstream folding toggle:
https://github.com/apache/arrow-rs/pull/11892 (issue:
https://github.com/apache/arrow-rs/issues/11891).
##########
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:
Thank you, this was covered in commit `test: document bloom folding after
writes`.
##########
native/proto/src/proto/operator.proto:
##########
@@ -679,6 +679,19 @@ message IcebergParquetWriteSettings {
// String written into parquet file metadata. JVM-side default is
// `"Apache Iceberg <ver> (Comet)"`.
string created_by = 7;
+ // Physical Parquet leaf paths for Iceberg columns whose
+ // `write.parquet.bloom-filter-enabled.column.<col>` property resolves to
true. The JVM driver
+ // translates Iceberg logical names to the actual Parquet paths used by
list/map encodings.
+ repeated string bloom_filter_enabled_columns = 8;
+ // Iceberg `write.parquet.bloom-filter-max-bytes` (default 1 MiB). Native
eligibility only
+ // admits powers of two in [32, 128 MiB], which parquet-rs can represent
exactly.
+ uint64 bloom_filter_max_bytes = 9;
+ // Effective per-column FPP, including Iceberg's 0.01 default. A value is
present for every
+ // enabled column so parquet-rs's different 0.05 default can never leak into
Iceberg writes.
+ map<string, double> bloom_filter_fpp_by_column = 10;
+ // User-provided Iceberg NDV values. Absence is significant: parquet-mr
allocates the full
+ // max in that case, whereas an explicit NDV sizes the filter before
applying the max as a cap.
+ map<string, uint64> bloom_filter_ndv_by_column = 11;
Review Comment:
Thank you, this was addressed in commit `refactor: group Iceberg bloom
column settings`.
##########
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:
Thank you, these runtime-specific paths were addressed in commit `fix: match
Iceberg bloom path resolution by version`.
##########
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:
Thank you, this was addressed in commits `feat: make native Iceberg bloom
folding opt-in` and `docs: group Bloom write prerequisites under accepted
divergences`.
##########
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:
Thank you, this was fixed in commit `fix: integrate Bloom write opt-in with
current Iceberg writer`.
--
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]
