diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index 49f35009202..f220e4f5238 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -154,7 +154,9 @@ A write is eligible only when ALL of the following hold: | `write.parquet.page-version` | unset or `v1` | | `write.parquet.shred-variants` | unset or `false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write) | | `write.parquet.variant-inference-buffer-size` | any value (only meaningful when shredding, which is gated) | -| `write.parquet.bloom-filter-enabled.column.` | unset or `false` | +| `write.parquet.bloom-filter-enabled.column.` | `true` or `false`; an explicit NDV enables the column even when this value is `false`, matching Iceberg's property application order | +| `write.parquet.bloom-filter-fpp.column.` / `write.parquet.bloom-filter-ndv.column.` | For every column named by an `enabled` property, FPP must be a finite double strictly between 0 and 1 and NDV must be a positive Java long no greater than `Long.MAX_VALUE / 8`; the Iceberg FPP default is `0.01` | +| `write.parquet.bloom-filter-max-bytes` | unset (Iceberg default: 1 MiB), or a power of two from 32 bytes through 128 MiB inclusive; every other value falls back to iceberg-java. A 32-byte value also falls back when explicit NDV/FPP would request a larger filter, because Parquet Java ignores that boundary as a maximum. | | `write.metadata.metrics.*` | any value (manifest metrics are re-derived on the JVM with Iceberg's own logic) | | `write.spark.fanout.enabled` | any value (the native writer implements both clustered and fanout modes) | | `write.target-file-size-bytes` | any value (file rolling cadence differs; see accepted divergences) | @@ -164,7 +166,7 @@ A write is eligible only when ALL of the following hold: Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` — everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g. -`bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions), +`stats-enabled.column.*`, keys added by future Iceberg versions), any `parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*` key in the session Hadoop configuration (with `HadoopFileIO`-backed output those reach iceberg-java's writer but not the native one). Also gated explicitly: any `encryption.*` key, @@ -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 +non-adaptive Iceberg path keeps its initial allocation. The native result can consequently use +less file space while remaining safe for every Parquet reader. + ### Physical file layout only (cosmetic) No Iceberg reader bases a planning or correctness decision on these; they change the bytes of diff --git a/native/core/src/execution/operators/iceberg_write.rs b/native/core/src/execution/operators/iceberg_write.rs index f7275ff1957..f1af0510570 100644 --- a/native/core/src/execution/operators/iceberg_write.rs +++ b/native/core/src/execution/operators/iceberg_write.rs @@ -64,6 +64,7 @@ use iceberg::writer::partitioning::unpartitioned_writer::UnpartitionedWriter; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel}; use parquet::file::properties::{EnabledStatistics, WriterProperties}; +use parquet::schema::types::ColumnPath; use datafusion_comet_proto::spark_operator::{ CompressionCodec as ProtoCompressionCodec, IcebergParquetWriteSettings, IcebergWrite, @@ -654,7 +655,7 @@ fn build_output_batch(manifest_bytes: Vec, output_schema: &SchemaRef) -> DFR /// the JVM re-derives from the footer with Iceberg's own `MetricsConfig` logic before commit. fn build_writer_properties(settings: &IcebergParquetWriteSettings) -> DFResult { let compression = compression_from_proto(settings.compression, settings.compression_level)?; - Ok(WriterProperties::builder() + let mut builder = WriterProperties::builder() .set_compression(compression) .set_created_by(settings.created_by.clone()) .set_max_row_group_bytes(Some(settings.row_group_size_bytes as usize)) @@ -665,8 +666,146 @@ fn build_writer_properties(settings: &IcebergParquetWriteSettings) -> DFResult ColumnPath { + ColumnPath::from(path.split('.').map(str::to_owned).collect::>()) +} + +// Match Apache Parquet Java's BlockSplitBloomFilter implementation bounds: +// https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L40-L50 +const BLOOM_FILTER_MIN_BYTES: usize = 32; +const BLOOM_FILTER_MAX_BYTES: usize = 128 * 1024 * 1024; +const BLOOM_FILTER_HASH_PROBES: f64 = 8.0; +const ICEBERG_DEFAULT_BLOOM_FILTER_FPP: f64 = 0.01; +#[cfg(test)] +const ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES: usize = 1024 * 1024; + +/// The positive denominator obtained by solving the Bloom-filter false-positive equation +/// `fpp = (1 - exp(-k * ndv / bits))^k` for `bits`, with the Parquet SBBF's `k = 8` probes. +/// +/// See the Apache Arrow Rust `parquet` implementation and its cited paper: +/// https://github.com/apache/arrow-rs/blob/58.4.0/parquet/src/bloom_filter/mod.rs#L369-L376 +/// http://algo2.iti.kit.edu/documents/cacheefficientbloomfilters-jea.pdf +fn bloom_filter_fpp_denominator(fpp: f64) -> f64 { + -(1.0 - fpp.powf(1.0 / BLOOM_FILTER_HASH_PROBES)).ln() +} + +/// Reproduce parquet-mr's non-adaptive allocation decision before translating the resulting +/// power-of-two byte size into parquet-rs's NDV-shaped API. An absent NDV requests the full cap; +/// an explicit NDV sizes from NDV/FPP and then applies the cap. The native eligibility gate only +/// admits representable power-of-two caps. +/// +/// Apache Parquet Java implementation: +/// https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L277-L301 +/// https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L195-L218 +fn parquet_mr_bloom_filter_bytes(ndv: Option, fpp: f64, max_bytes: usize) -> usize { + let Some(ndv) = ndv else { + return max_bytes; + }; + + let calculated = BLOOM_FILTER_HASH_PROBES * ndv as f64 / bloom_filter_fpp_denominator(fpp); + let mut num_bits = calculated as i32; + let upper_bits = (BLOOM_FILTER_MAX_BYTES * 8) as i32; + if num_bits > upper_bits || calculated < 0.0 { + num_bits = upper_bits; + } + // This deliberately mirrors parquet-mr 1.17's integer expression, including its unusual + // mask, so allocation thresholds remain compatible rather than merely mathematically close. + num_bits = (num_bits + 255) & !256; + num_bits = num_bits.max((BLOOM_FILTER_MIN_BYTES * 8) as i32); + let requested = (num_bits as usize) / 8; + let allocated = requested + .clamp(BLOOM_FILTER_MIN_BYTES, BLOOM_FILTER_MAX_BYTES) + .next_power_of_two(); + // The eligibility gate excludes max=32 when this cap would change parquet-mr's allocation. + allocated.min(max_bytes) +} + +/// Mirror the NDV/FPP sizing and power-of-two allocation used by the Apache Arrow Rust `parquet` +/// crate. Its source derives the formula from the standard Bloom-filter false-positive equation +/// with eight hash probes and links the underlying cache-efficient Bloom-filter paper: +/// https://github.com/apache/arrow-rs/blob/58.4.0/parquet/src/bloom_filter/mod.rs#L363-L395 +fn parquet_rs_bloom_filter_bytes(ndv: u64, fpp: f64) -> usize { + let num_bits = + (BLOOM_FILTER_HASH_PROBES * ndv as f64 / bloom_filter_fpp_denominator(fpp)) as usize; + (num_bits / 8) + .clamp(BLOOM_FILTER_MIN_BYTES, BLOOM_FILTER_MAX_BYTES) + .next_power_of_two() +} + +/// Encode an exact power-of-two allocation using parquet-rs 58.x's public NDV/FPP setters. +/// +/// A target `B > 32` is selected by every raw byte count in `(B/2, B]`. Aim at `3B/4`, far from +/// either floating-point boundary, and verify using the exact parquet-rs sizing expression. The +/// binary-search fallback covers unusual but still representable FPP values without relying on +/// the inverse formula landing on a particular floating-point integer. +fn synthetic_ndv_for_bloom_filter_bytes(target_bytes: usize, fpp: f64) -> DFResult { + let fpp_denominator = bloom_filter_fpp_denominator(fpp); + // Any raw size in (B / 2, B] rounds up to the target power-of-two allocation B. Choose the + // midpoint of that interval to stay away from floating-point boundaries at either end. + let raw_target_bytes = target_bytes as f64 * 3.0 / 4.0; + let candidate = ((raw_target_bytes * fpp_denominator).round() as u64).max(1); + if parquet_rs_bloom_filter_bytes(candidate, fpp) == target_bytes { + return Ok(candidate); + } + + // Rust's standard binary-search helpers operate on materialized slices; this is a lower-bound + // search over the implicit NDV domain `1..=u64::MAX`, so keep the numeric search explicit. + let mut low = 1_u64; + let mut high = u64::MAX; + while low < high { + let mid = low + (high - low) / 2; + if parquet_rs_bloom_filter_bytes(mid, fpp) < target_bytes { + low = mid.saturating_add(1); + } else { + high = mid; + } + } + if parquet_rs_bloom_filter_bytes(low, fpp) == target_bytes { + Ok(low) + } else { + Err(DataFusionError::Internal(format!( + "FPP {fpp} cannot represent a {target_bytes}-byte parquet-rs Bloom filter" + ))) + } +} + +fn validate_bloom_filter_inputs(fpp: f64, bytes: usize) -> DFResult<()> { + if !fpp.is_finite() || !(0.0..1.0).contains(&fpp) { + return Err(DataFusionError::Internal(format!( + "Bloom filter FPP must be finite and strictly between 0 and 1, got {fpp}" + ))); + } + if !(BLOOM_FILTER_MIN_BYTES..=BLOOM_FILTER_MAX_BYTES).contains(&bytes) + || !bytes.is_power_of_two() + { + return Err(DataFusionError::Internal(format!( + "Bloom filter byte size must be a power of two in [{BLOOM_FILTER_MIN_BYTES}, {BLOOM_FILTER_MAX_BYTES}], got {bytes}" + ))); + } + Ok(()) } fn compression_from_proto(codec: i32, level: Option) -> DFResult { @@ -725,6 +864,10 @@ mod tests { dict_size_bytes: 2 * 1024 * 1024, page_row_limit: 20_000, created_by: "Apache Iceberg (Comet test)".to_string(), + bloom_filter_enabled_columns: Vec::new(), + bloom_filter_max_bytes: ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES as u64, + bloom_filter_fpp_by_column: Default::default(), + bloom_filter_ndv_by_column: Default::default(), } } @@ -804,6 +947,107 @@ mod tests { assert_eq!(props.created_by(), "Apache Iceberg 1.7.1 (Comet 0.16.0)"); } + #[test] + fn enables_bloom_filters_only_for_configured_columns() { + let mut settings = base_settings(); + settings.bloom_filter_enabled_columns = vec![ + "id".to_string(), + "tags.list.element".to_string(), + "attrs.key_value.value".to_string(), + ]; + let props = build_writer_properties(&settings).unwrap(); + assert!(props + .bloom_filter_properties(&parquet_column_path("id")) + .is_some()); + assert!(props + .bloom_filter_properties(&parquet_column_path("tags.list.element")) + .is_some()); + assert!(props + .bloom_filter_properties(&parquet_column_path("attrs.key_value.value")) + .is_some()); + assert!(props + .bloom_filter_properties(&parquet_column_path("other")) + .is_none()); + } + + #[test] + fn bloom_filter_defaults_match_iceberg() { + // Iceberg's documented write-property defaults: + // https://iceberg.apache.org/docs/latest/configuration/#write-properties + let mut settings = base_settings(); + settings.bloom_filter_enabled_columns = vec!["id".to_string()]; + let props = build_writer_properties(&settings).unwrap(); + let bloom = props.bloom_filter_properties(&"id".into()).unwrap(); + assert_eq!(bloom.fpp(), ICEBERG_DEFAULT_BLOOM_FILTER_FPP); + assert_eq!( + parquet_rs_bloom_filter_bytes(bloom.ndv(), bloom.fpp()), + ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES + ); + } + + #[test] + fn synthetic_ndv_hits_every_supported_size_away_from_float_boundaries() { + for fpp in [0.0001, ICEBERG_DEFAULT_BLOOM_FILTER_FPP, 0.05, 0.5, 0.99] { + let mut bytes = BLOOM_FILTER_MIN_BYTES; + while bytes <= BLOOM_FILTER_MAX_BYTES { + let ndv = synthetic_ndv_for_bloom_filter_bytes(bytes, fpp).unwrap(); + assert_eq!(parquet_rs_bloom_filter_bytes(ndv, fpp), bytes); + bytes *= 2; + } + } + } + + #[test] + fn explicit_ndv_controls_requested_size_and_max_only_caps_it() { + let small = parquet_mr_bloom_filter_bytes( + Some(10), + ICEBERG_DEFAULT_BLOOM_FILTER_FPP, + 64 * 1024 * 1024, + ); + assert!(small < 64 * 1024 * 1024); + + let capped = parquet_mr_bloom_filter_bytes( + Some(100_000_000), + ICEBERG_DEFAULT_BLOOM_FILTER_FPP, + ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES, + ); + assert_eq!(capped, ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES); + let uncapped = + parquet_mr_bloom_filter_bytes(None, ICEBERG_DEFAULT_BLOOM_FILTER_FPP, 64 * 1024 * 1024); + assert_eq!(uncapped, 64 * 1024 * 1024); + } + + #[test] + fn minimum_bloom_filter_size_is_representable() { + assert_eq!( + parquet_mr_bloom_filter_bytes( + Some(1), + ICEBERG_DEFAULT_BLOOM_FILTER_FPP, + BLOOM_FILTER_MIN_BYTES, + ), + BLOOM_FILTER_MIN_BYTES + ); + let ndv = synthetic_ndv_for_bloom_filter_bytes( + BLOOM_FILTER_MIN_BYTES, + ICEBERG_DEFAULT_BLOOM_FILTER_FPP, + ) + .unwrap(); + assert_eq!( + parquet_rs_bloom_filter_bytes(ndv, ICEBERG_DEFAULT_BLOOM_FILTER_FPP), + BLOOM_FILTER_MIN_BYTES + ); + } + + #[test] + fn impossible_synthetic_ndv_is_rejected() { + let err = synthetic_ndv_for_bloom_filter_bytes( + ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES, + f64::MIN_POSITIVE, + ) + .unwrap_err(); + assert!(format!("{err}").contains("cannot represent")); + } + #[test] fn rejects_unknown_codec() { let mut settings = base_settings(); @@ -906,6 +1150,10 @@ mod tests { dict_size_bytes: 2 * 1024 * 1024, page_row_limit: 20_000, created_by: "Apache Iceberg (Comet integration test)".to_string(), + bloom_filter_enabled_columns: Vec::new(), + bloom_filter_max_bytes: ICEBERG_DEFAULT_BLOOM_FILTER_MAX_BYTES as u64, + bloom_filter_fpp_by_column: Default::default(), + bloom_filter_ndv_by_column: Default::default(), }; Arc::new(IcebergWriteCommon { catalog_properties: HashMap::new(), diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 7b34f84012e..3f4d30f8a12 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -679,6 +679,19 @@ message IcebergParquetWriteSettings { // String written into parquet file metadata. JVM-side default is // `"Apache Iceberg (Comet)"`. string created_by = 7; + // Physical Parquet leaf paths for Iceberg columns whose + // `write.parquet.bloom-filter-enabled.column.` 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 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 bloom_filter_ndv_by_column = 11; } // Broadcast payload -- one of these per write, identical for every task. diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index fb705ce14b4..86ffcb2a503 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -37,6 +37,10 @@ import org.apache.comet.util.ClassLoaders */ object IcebergReflection extends Logging { + case class ParquetPathResolution( + pathByIcebergColumnName: Map[String, String], + renamedIcebergColumnNames: Set[String]) + /** * Iceberg class names used throughout Comet. */ @@ -56,6 +60,7 @@ object IcebergReflection extends Logging { val SPARK_BATCH_QUERY_SCAN = "org.apache.iceberg.spark.source.SparkBatchQueryScan" val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan" val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil" + val PARQUET_SCHEMA_UTIL = "org.apache.iceberg.parquet.ParquetSchemaUtil" val TABLE = "org.apache.iceberg.Table" val PARTITIONING = "org.apache.iceberg.Partitioning" val SPARK_WRITE = "org.apache.iceberg.spark.source.SparkWrite" @@ -580,6 +585,74 @@ object IcebergReflection extends Logging { } } + // scalastyle:off line.size.limit + /** + * Maps Iceberg's logical primitive-column names to their physical Parquet leaf paths. + * + * This mirrors the map Iceberg Java builds before applying per-column writer settings. In + * particular, Parquet's canonical three-level encodings insert `list` for array elements and + * `key_value` for map keys/values, so logical names such as `tags.element` and `attrs.value` + * cannot be passed directly to Parquet writer properties. + * + * See: + * https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java#L411-L421 + */ + // scalastyle:on line.size.limit + def getParquetPathResolution(schema: Any): Option[ParquetPathResolution] = { + import scala.jdk.CollectionConverters._ + try { + val parquetSchemaUtil = loadClass(ClassNames.PARQUET_SCHEMA_UTIL) + val parquetSchema = parquetSchemaUtil + .getMethod("convert", loadClass(ClassNames.SCHEMA), classOf[String]) + .invoke(null, schema.asInstanceOf[AnyRef], "table") + val columns = getMethod(parquetSchema.getClass, "getColumns") + .invoke(parquetSchema) + .asInstanceOf[java.util.List[AnyRef]] + val findColumnName = getMethod(schema.getClass, "findColumnName", classOf[Int]) + val findField = getMethod(schema.getClass, "findField", classOf[Int]) + + val resolved = columns.asScala.flatMap { column => + val primitiveType = getMethod(column.getClass, "getPrimitiveType").invoke(column) + val parquetId = getMethod(primitiveType.getClass, "getId").invoke(primitiveType) + Option(parquetId).flatMap { id => + val fieldId = getMethod(id.getClass, "intValue").invoke(id).asInstanceOf[Int] + Option(findColumnName.invoke(schema, Int.box(fieldId))).map { icebergName => + val parquetPathParts = getMethod(column.getClass, "getPath") + .invoke(column) + .asInstanceOf[Array[String]] + var parquetType = parquetSchema.asInstanceOf[AnyRef] + val renamed = parquetPathParts.exists { pathPart => + parquetType = getMethod(parquetType.getClass, "getType", classOf[String]) + .invoke(parquetType, pathPart) + .asInstanceOf[AnyRef] + Option(getMethod(parquetType.getClass, "getId").invoke(parquetType)).exists { + parquetFieldId => + val idValue = getMethod(parquetFieldId.getClass, "intValue") + .invoke(parquetFieldId) + .asInstanceOf[Int] + Option(findField.invoke(schema, Int.box(idValue))).exists { icebergField => + val icebergFieldName = getMethod(icebergField.getClass, "name") + .invoke(icebergField) + .asInstanceOf[String] + icebergFieldName != pathPart + } + } + } + (icebergName.asInstanceOf[String], parquetPathParts.mkString("."), renamed) + } + } + }.toSeq + Some( + ParquetPathResolution( + resolved.map { case (name, path, _) => name -> path }.toMap, + resolved.collect { case (name, _, true) => name }.toSet)) + } catch { + case e: Exception => + logError(s"Iceberg reflection failure: Parquet column paths: ${e.getMessage}") + None + } + } + /** * All schema versions a table has had (table.schemas().values()), for resolving field ids of * columns that have since been dropped -- mirrors Iceberg-Java's FieldLookup. table.schemas() @@ -1289,6 +1362,12 @@ object IcebergReflection extends Logging { def tablePropertyIntConstant(fieldName: String): Int = readTablePropertiesField(fieldName).asInstanceOf[Integer].intValue() + def tablePropertyDoubleConstantOpt(fieldName: String): Option[Double] = + tablePropertiesClassOpt.flatMap { cls => + try Some(cls.getField(fieldName).get(null).asInstanceOf[java.lang.Double].doubleValue()) + catch { case _: NoSuchFieldException => None } + } + /** * Like [[tablePropertyConstant]] but returns `None` when the constant is absent in the Iceberg * version on the classpath rather than throwing. Used to gate behaviour that only some Iceberg diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala index 4aee4123ff3..0ca47264fdc 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala @@ -49,6 +49,10 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { IcebergReflection.tablePropertyConstant("WRITE_LOCATION_PROVIDER_IMPL") lazy val BloomFilterColumnEnabledPrefix: String = IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX") + lazy val ParquetBloomFilterMaxBytes: String = + IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_MAX_BYTES") + val ParquetBloomFilterColumnFppPrefix = "write.parquet.bloom-filter-fpp.column." + val ParquetBloomFilterColumnNdvPrefix = "write.parquet.bloom-filter-ndv.column." lazy val ParquetRowGroupCheckMinRecordCount: String = IcebergReflection.tablePropertyConstant("PARQUET_ROW_GROUP_CHECK_MIN_RECORD_COUNT") lazy val ParquetRowGroupCheckMinRecordCountDefault: Int = @@ -111,10 +115,14 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { PropertyKeys.ParquetRowGroupCheckMaxRecordCount, PropertyKeys.ParquetPageVersion, PropertyKeys.ParquetShredVariants, - PropertyKeys.ParquetVariantBufferSize) + PropertyKeys.ParquetVariantBufferSize, + PropertyKeys.ParquetBloomFilterMaxBytes) private lazy val vettedParquetWritePrefixes: Seq[String] = - Seq(PropertyKeys.BloomFilterColumnEnabledPrefix) + Seq( + PropertyKeys.BloomFilterColumnEnabledPrefix, + PropertyKeys.ParquetBloomFilterColumnFppPrefix, + PropertyKeys.ParquetBloomFilterColumnNdvPrefix) override def getSupportLevel(op: IcebergWriteExec): SupportLevel = try { @@ -175,12 +183,12 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { requireFormatVersionAtMostTwo, requireNoUuidColumns, requireNoEncryptionPrefix, - requireNoBloomFilterColumnsEnabled, requireRowGroupCheckMinRecordCountAtDefault, requireRowGroupCheckMaxRecordCountAtDefault, requireParquetPageVersionDefault, requireShredVariantsDisabled, requireNativeSupportedCompressionLevel, + requireNativeSupportedBloomFilterProperties, requireOnlyVettedParquetWriteProperties, requirePropertyAbsent( PropertyKeys.ParquetEnableDictionary, @@ -252,13 +260,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { // `CometIcebergWriteExec`), so every `write.metadata.metrics.*` value behaves exactly as it // does on the iceberg-java path. - private val requireNoBloomFilterColumnsEnabled: TriggerRule = ctx => { - val prefix = PropertyKeys.BloomFilterColumnEnabledPrefix - ctx.properties - .find { case (k, v) => k.startsWith(prefix) && v.equalsIgnoreCase("true") } - .map { case (k, _) => s"$k=true: bloom filters unsupported" } - } - private val requireParquetPageVersionDefault: TriggerRule = ctx => { val key = PropertyKeys.ParquetPageVersion ctx.properties @@ -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 = + properties.get(PropertyKeys.ParquetBloomFilterMaxBytes).flatMap { raw => + scala.util.Try(java.lang.Integer.parseInt(raw)).toOption match { + case None => Some(s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw is not a Java int") + case Some(value) + if value < MinBloomFilterBytes || value > MaxBloomFilterBytes || + (value & (value - 1)) != 0 => + Some( + s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$value must be a power of two " + + s"in [$MinBloomFilterBytes, $MaxBloomFilterBytes] for native writes") + case Some(_) => None + } + } + + maxRejection.orElse { + val maxBytes = properties + .get(PropertyKeys.ParquetBloomFilterMaxBytes) + .flatMap(raw => scala.util.Try(java.lang.Integer.parseInt(raw)).toOption) + .getOrElse(IcebergWriteProtoTranslation.Defaults.BloomFilterMaxBytes) + // Iceberg visits every enabled-prefix entry and applies enabled, FPP, then NDV. Validate + // the associated shape properties even for enabled=false; a valid NDV also re-enables the + // filter in parquet-mr. + val configured = properties.iterator.collect { + case (key, _) if key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) => + key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length) + }.toSeq + configured.iterator + .flatMap { column => + val fppKey = PropertyKeys.ParquetBloomFilterColumnFppPrefix + column + val ndvKey = PropertyKeys.ParquetBloomFilterColumnNdvPrefix + column + val parsedFpp = properties.get(fppKey) match { + case Some(raw) => scala.util.Try(java.lang.Double.parseDouble(raw)).toOption + case None => Some(IcebergWriteProtoTranslation.Defaults.BloomFilterFpp) + } + val parsedNdv = properties + .get(ndvKey) + .flatMap(raw => scala.util.Try(java.lang.Long.parseLong(raw)).toOption) + val fppError = properties.get(fppKey).flatMap { raw => + parsedFpp match { + case Some(value) + if value > 0.0d && value < 1.0d && java.lang.Double.isFinite(value) && + bloomFilterSizesRepresentable(maxBytes, value) => + None + case Some(value) + if value > 0.0d && value < 1.0d && java.lang.Double.isFinite(value) => + Some(s"$fppKey=$raw cannot represent the configured native Bloom sizes") + case _ => Some(s"$fppKey=$raw must be a finite double strictly between 0 and 1") + } + } + val ndvError = properties.get(ndvKey).flatMap { raw => + parsedNdv match { + case Some(value) if value > 0L && value <= MaxNonOverflowingBloomFilterNdv => None + case Some(value) if value > MaxNonOverflowingBloomFilterNdv => + Some(s"$ndvKey=$raw exceeds $MaxNonOverflowingBloomFilterNdv; " + + "parquet-mr Bloom sizing may overflow") + case _ => Some(s"$ndvKey=$raw must be a positive Java long") + } + } + val ignoredMinimumCapError = parsedNdv.collect { + case ndv + if maxBytes == MinBloomFilterBytes && ndv > 0L && + ndv <= MaxNonOverflowingBloomFilterNdv && + parsedFpp.exists( + parquetMrRequestedBloomFilterBytes(ndv, _) > MinBloomFilterBytes) => + s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$MinBloomFilterBytes is ignored by " + + s"parquet-mr for $ndvKey=$ndv" + } + Seq(fppError, ndvError, ignoredMinimumCapError).flatten + } + .toSeq + .headOption + } + } + + // Planning-time counterpart of the native inverse-NDV check. A target B is safely encoded by + // aiming at 3B/4, in the interior of parquet-rs's (B/2, B] round-up interval. Requiring every + // power-of-two through the configured cap is conservative and keeps pathological-but-valid + // floating-point FPPs on the JVM path rather than discovering them after task launch. + private def bloomFilterSizesRepresentable(maxBytes: Int, fpp: Double): Boolean = { + val denominator = + -Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble)) + if (!java.lang.Double.isFinite(denominator) || denominator <= 0.0d) return false + + Iterator + .iterate(MinBloomFilterBytes)(_ * 2) + .takeWhile(_ <= maxBytes) + .forall { target => + val ndv = Math.max(1L, Math.round(target.toDouble * 0.75d * denominator)) + val calculatedBits = (-BloomFilterHashProbes.toDouble * ndv.toDouble / + Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble))).toLong + val rawBytes = Math.max( + MinBloomFilterBytes.toLong, + Math.min(MaxBloomFilterBytes.toLong, calculatedBits / 8L)) + val allocated = java.lang.Long.highestOneBit(rawBytes - 1L) << 1 + allocated == target.toLong + } + } + + /** The uncapped byte count parquet-mr passes to its explicit-NDV constructor path. */ + private def parquetMrRequestedBloomFilterBytes(ndv: Long, fpp: Double): Int = { + // Keep the long multiplication before floating-point conversion to match parquet-mr: + // 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#L277-L301 + // scalastyle:on line.size.limit + val calculated = (-BloomFilterHashProbes.toLong * ndv).toDouble / + Math.log(1.0d - Math.pow(fpp, 1.0d / BloomFilterHashProbes.toDouble)) + val maxBits = MaxBloomFilterBytes << 3 + var bits = calculated.toInt + if (bits > maxBits || calculated < 0.0d) bits = maxBits + val bitsPerBlock = MinBloomFilterBytes << 3 + bits = (bits + bitsPerBlock - 1) & ~bitsPerBlock + Math.max(bits, bitsPerBlock) / 8 + } + private val requireOnlyVettedParquetWriteProperties: TriggerRule = ctx => ctx.properties .find { case (k, _) => @@ -626,9 +777,36 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { // `option("write-parquet-compression-codec", "gzip")`) survive into the native writer. val resolvedWriteProperties = IcebergReflection.getWritePropertiesFromSparkWrite(sparkWrite).getOrElse(Map.empty) - val effectiveProperties = properties ++ resolvedWriteProperties - val parquetSettings = - IcebergWriteProtoTranslation.buildParquetSettings(effectiveProperties, createdBy) + val effectiveProperties = + interpretedBloomFilterProperties(properties ++ resolvedWriteProperties) + val parquetPathByIcebergColumnName = + if (IcebergWriteProtoTranslation.hasEnabledBloomFilters(effectiveProperties)) { + val resolution = IcebergReflection.getParquetPathResolution(writeSchema).getOrElse { + withFallbackReason(op, "Could not resolve physical Parquet paths for Bloom filters") + return None + } + val renamedColumns = + IcebergWriteProtoTranslation + .enabledBloomFilterColumnNames(effectiveProperties) + .filter(resolution.renamedIcebergColumnNames) + if (renamedColumns.nonEmpty) { + // Iceberg Java sanitizes these Parquet names, while the pinned iceberg-rust Arrow + // conversion preserves them. Passing the Java path to parquet-rs would silently miss + // the native writer column, so retain the JVM writer until paths travel structurally. + withFallbackReason( + op, + "Bloom-filter columns are renamed in Iceberg Java's Parquet schema: " + + renamedColumns.mkString(", ")) + return None + } + resolution.pathByIcebergColumnName + } else { + Map.empty[String, String] + } + val parquetSettings = IcebergWriteProtoTranslation.buildParquetSettings( + effectiveProperties, + createdBy, + parquetPathByIcebergColumnName) // `FileIO.properties()` misses configuration a HadoopFileIO carries through the Hadoop // Configuration instead (fs.s3a.* credentials, custom endpoint, path-style access), which diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala b/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala index 531f3c227dd..108b6b38844 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala @@ -57,6 +57,15 @@ object IcebergWriteProtoTranslation { IcebergReflection.tablePropertyConstant("PARQUET_PAGE_ROW_LIMIT") lazy val ParquetDictSizeBytes: String = IcebergReflection.tablePropertyConstant("PARQUET_DICT_SIZE_BYTES") + lazy val ParquetBloomFilterColumnEnabledPrefix: String = + IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX") + lazy val ParquetBloomFilterMaxBytes: String = + IcebergReflection.tablePropertyConstant("PARQUET_BLOOM_FILTER_MAX_BYTES") + // These were added to Iceberg after bloom enablement/max-bytes. Literals let the translation + // support them without a hard binary dependency; the caller removes either prefix when the + // Iceberg runtime does not interpret it, matching that runtime's JVM writer. + val ParquetBloomFilterColumnFppPrefix = "write.parquet.bloom-filter-fpp.column." + val ParquetBloomFilterColumnNdvPrefix = "write.parquet.bloom-filter-ndv.column." } /** Iceberg's numeric defaults, pulled at runtime so they stay in lock-step with the runtime. */ @@ -69,6 +78,14 @@ object IcebergWriteProtoTranslation { IcebergReflection.tablePropertyIntConstant("PARQUET_DICT_SIZE_BYTES_DEFAULT").toLong lazy val PageRowLimit: Int = IcebergReflection.tablePropertyIntConstant("PARQUET_PAGE_ROW_LIMIT_DEFAULT") + lazy val BloomFilterMaxBytes: Int = + IcebergReflection.tablePropertyIntConstant("PARQUET_BLOOM_FILTER_MAX_BYTES_DEFAULT") + // Iceberg introduced the public constant together with the FPP property. Keep the literal + // fallback for runtimes old enough not to expose it; 0.01 is also parquet-mr's default. + lazy val BloomFilterFpp: Double = + IcebergReflection + .tablePropertyDoubleConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_DEFAULT") + .getOrElse(0.01d) } /** @@ -100,10 +117,43 @@ object IcebergWriteProtoTranslation { } } + private def configuredBloomFilterColumnNames(props: Map[String, String]): Seq[String] = + props.iterator + .collect { + case (key, _) if key.startsWith(Keys.ParquetBloomFilterColumnEnabledPrefix) => + key.substring(Keys.ParquetBloomFilterColumnEnabledPrefix.length) + } + .toSeq + .sorted + + private[operator] def enabledBloomFilterColumnNames(props: Map[String, String]): Seq[String] = + configuredBloomFilterColumnNames(props).filter { column => + val enabled = + java.lang.Boolean.valueOf(props(Keys.ParquetBloomFilterColumnEnabledPrefix + column)) + // Iceberg applies enabled, FPP, and NDV in that order. parquet-mr's NDV setter enables the + // column, so an explicit NDV wins over enabled=false. + enabled || props.contains(Keys.ParquetBloomFilterColumnNdvPrefix + column) + } + + def hasEnabledBloomFilters(props: Map[String, String]): Boolean = + enabledBloomFilterColumnNames(props).nonEmpty + + /** + * Test convenience for schemas where Iceberg logical names and physical Parquet paths are + * identical. Production translation must supply Iceberg Java's logical-to-physical path map. + */ + private[operator] def buildParquetSettings( + props: Map[String, String], + createdBy: String): IcebergParquetWriteSettings = { + val identityPaths = enabledBloomFilterColumnNames(props).map(name => name -> name).toMap + buildParquetSettings(props, createdBy, identityPaths) + } + /** Builds the parquet settings message. Pure: no SparkWrite or Iceberg `Table` access. */ def buildParquetSettings( props: Map[String, String], - createdBy: String): IcebergParquetWriteSettings = { + createdBy: String, + parquetPathByIcebergColumnName: Map[String, String]): IcebergParquetWriteSettings = { val rowGroupSize = parseJavaInt(props, Keys.ParquetRowGroupSizeBytes, Defaults.RowGroupSizeBytes.toInt).toLong val pageSize = @@ -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 -> _) + } + .sortBy(_._2) + val bloomFilterEnabledColumns = bloomFilterColumns.map(_._2) + val bloomFilterMaxBytes = + parseJavaInt(props, Keys.ParquetBloomFilterMaxBytes, Defaults.BloomFilterMaxBytes).toLong + val bloomFilterFppByColumn = bloomFilterColumns.map { case (icebergName, parquetPath) => + val value = props + .get(Keys.ParquetBloomFilterColumnFppPrefix + icebergName) + .map(java.lang.Double.parseDouble) + .getOrElse(Defaults.BloomFilterFpp) + parquetPath -> value + }.toMap + val bloomFilterNdvByColumn = bloomFilterColumns.flatMap { case (icebergName, parquetPath) => + props + .get(Keys.ParquetBloomFilterColumnNdvPrefix + icebergName) + .map(value => parquetPath -> java.lang.Long.parseLong(value)) + }.toMap val builder = IcebergParquetWriteSettings .newBuilder() .setCompression(compression) @@ -120,6 +192,14 @@ object IcebergWriteProtoTranslation { .setDictSizeBytes(dictSize) .setPageRowLimit(pageRowLimit) .setCreatedBy(createdBy) + .addAllBloomFilterEnabledColumns(bloomFilterEnabledColumns.asJava) + .setBloomFilterMaxBytes(bloomFilterMaxBytes) + .putAllBloomFilterFppByColumn(bloomFilterFppByColumn.map { case (k, v) => + k -> Double.box(v) + }.asJava) + .putAllBloomFilterNdvByColumn(bloomFilterNdvByColumn.map { case (k, v) => + k -> Long.box(v) + }.asJava) resolveCompressionLevel(props, compression).foreach(builder.setCompressionLevel) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 8095e6feb05..17477ae4dff 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -19,14 +19,18 @@ package org.apache.comet -import java.io.File +import java.io.{ByteArrayOutputStream, File} import java.util.concurrent.{CountDownLatch, TimeUnit} import scala.collection.mutable import scala.concurrent.{Await, Future} import scala.concurrent.ExecutionContext.Implicits.global import scala.concurrent.duration.DurationInt +import scala.jdk.CollectionConverters._ +import org.apache.hadoop.fs.Path +import org.apache.parquet.hadoop.ParquetFileReader +import org.apache.parquet.hadoop.util.HadoopInputFile import org.apache.spark.SparkConf import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.Row @@ -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, + attrs MAP + ) 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 + // Java retains its initial maximum allocation. Folding is safe for readers, so require + // only that native uses no more space and still contains every inserted value. + assert(native.zip(jvm).forall { case (left, right) => left.length <= right.length }) + assertParquetBloomContainsInts(nativeTable, "id", ids) + } + if (suffix == "binding_cap") { + assert(native.forall(_.length == 64), s"expected binding 64-byte cap for $nativeTable") + assertParquetBloomContainsInts(nativeTable, "id", ids) + } + } + } + } + + test("explicit underestimated NDV retains parquet-mr allocation precedence") { + assumeNativeAcceleration() + assumeIcebergBloomShapeProperties() + withIcebergCatalog { warehouseDir => + // max-bytes is only a cap in parquet-mr; it does not enlarge a filter whose explicit NDV + // was underestimated. This comparison prevents Comet from silently replacing the user's + // NDV with an artificial NDV derived from the much larger maximum. + val properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='true', " + + "'write.parquet.bloom-filter-fpp.column.id'='0.01', " + + "'write.parquet.bloom-filter-ndv.column.id'='10', " + + "'write.parquet.bloom-filter-max-bytes'='67108864'") + createTable(warehouseDir, "bloom_low_ndv_native", partitionSpec = "", properties) + createTable(warehouseDir, "bloom_low_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) FROM range(0, 4096, 1, 1)") + + assertNativeWriteEngages("bloom_low_ndv_native", 0 until 4096) { + insert("bloom_low_ndv_native") + } + insert("bloom_low_ndv_jvm") + + val native = parquetBloomFilterBytes("bloom_low_ndv_native", "id") + val jvm = parquetBloomFilterBytes("bloom_low_ndv_jvm", "id") + assert(native.map(_.length) == jvm.map(_.length)) + assert(native.zip(jvm).forall { case (left, right) => + java.util.Arrays.equals(left, right) + }) + } + } + + test("large representable max folds natively while adjacent non-power-of-two falls back") { + assumeNativeAcceleration() + withIcebergCatalog { warehouseDir => + val powerOfTwo = 64 * 1024 * 1024 + val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true', " + createTable( + warehouseDir, + "bloom_large_native", + partitionSpec = "", + properties = Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='$powerOfTwo'")) + createTable( + warehouseDir, + "bloom_large_jvm", + partitionSpec = "", + properties = + Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='${powerOfTwo + 1}'")) + + 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, 256, 1, 1)") + + assertNativeWriteEngages("bloom_large_native", 0 until 256) { + insert("bloom_large_native") + } + assertNativeWriteDoesNotEngage("bloom_large_jvm", 0 until 256) { + insert("bloom_large_jvm") + } + + val nativeBytes = parquetBloomFilterBytes("bloom_large_native", "id") + val nativeSize = nativeBytes.head.length + val jvmSize = parquetBloomFilterBytes("bloom_large_jvm", "id").head.length + assert(nativeSize < 1024 * 1024, s"expected folding, got $nativeSize bytes") + // Iceberg runtimes bundle different Parquet Java versions: newer versions retain the exact + // cap while older ones round it upward. Both demonstrate that the fallback avoids folding + // this deliberately oversized filter down to the native allocation. + assert(jvmSize > powerOfTwo, s"expected an oversized JVM filter, got $jvmSize bytes") + assertParquetBloomContainsInts("bloom_large_native", "id", 0 until 256) + + // The adjacent non-power-of-two JVM filter is intentionally much larger, so comparing its + // bytes with the folded native filter would be meaningless. Instead, write the same values + // through the JVM writer with a cap equal to the observed folded size. Power-of-two folding + // must produce exactly the same SBBF bitset as hashing directly into that final allocation. + createTable( + warehouseDir, + "bloom_large_folded_jvm", + partitionSpec = "", + properties = Some(enabled + s"'write.parquet.bloom-filter-max-bytes'='$nativeSize'")) + withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false") { + insert("bloom_large_folded_jvm") + } + assertRows("bloom_large_folded_jvm", 0 until 256) + val foldedJvmBytes = parquetBloomFilterBytes("bloom_large_folded_jvm", "id") + assert(nativeBytes.map(_.length) == foldedJvmBytes.map(_.length)) + assert( + nativeBytes.zip(foldedJvmBytes).forall { case (left, right) => + java.util.Arrays.equals(left, right) + }, + "folded native SBBF must be byte-identical to a same-sized JVM SBBF") + } + } + + test("out-of-range bloom max uses the classic writer") { + assumeNativeAcceleration() + withIcebergCatalog { warehouseDir => + // Leave Bloom filters disabled so the stock writer can demonstrate planning fallback + // without allocating a 128 MiB filter for the oversized case. + Seq("31", "134217729").zipWithIndex.foreach { case (max, index) => + val table = s"bloom_range_fallback_$index" + createTable( + warehouseDir, + table, + partitionSpec = "", + properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$max'")) + assertNativeWriteDoesNotEngage(table, Seq(index)) { + spark.sql(s"INSERT INTO cat.db.$table VALUES ($index, 'region', 1.0)") + } + } + } + } + // What Iceberg's Spark writer stamps for `sort_order_id` on appended files changed across // releases: through 1.10 `SparkWrite$WriterFactory` never wires the table sort order (files // get 0 even on a sorted table); 1.11 added `SparkWriteConf.outputSortOrderId` and stamps the @@ -1806,11 +2257,148 @@ class CometIcebergWriteActionSuite assert(ids == expectedIds, s"expected $expectedIds, got $ids") } + /** + * Reads every current data-file footer and verifies that parquet-rs emitted bloom-filter data + * exactly for the configured columns. Checking the plan alone would not catch a translation bug + * that selected `CometIcebergWriteExec` but silently omitted the bloom writer properties. + */ + private def assertParquetBloomFilters(tableName: String, enabledColumns: Set[String]): Unit = { + val paths = spark + .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files") + .collect() + .map(_.getString(0)) + .toSeq + assert(paths.nonEmpty, s"expected $tableName to have at least one data file") + + val conf = spark.sparkContext.hadoopConfiguration + paths.foreach { path => + val input = HadoopInputFile.fromPath(new Path(path.stripPrefix("file:")), conf) + val reader = ParquetFileReader.open(input) + try { + val columns = reader.getFooter.getBlocks.asScala.flatMap(_.getColumns.asScala) + assert(columns.nonEmpty, s"expected at least one column chunk in $path") + columns.foreach { column => + val columnName = column.getPath.toDotString + val bloomFilter = reader.readBloomFilter(column) + if (enabledColumns.contains(columnName)) { + assert( + bloomFilter != null && bloomFilter.getBitsetSize > 0, + s"expected a non-empty bloom filter for $columnName in $path") + } else { + assert( + bloomFilter == null, + s"expected no bloom filter for unconfigured column $columnName in $path") + } + } + } finally { + reader.close() + } + } + } + + private def parquetBloomFilterBytes(tableName: String, columnName: String): Seq[Array[Byte]] = { + val paths = spark + .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files ORDER BY file_path") + .collect() + .map(_.getString(0)) + val conf = spark.sparkContext.hadoopConfiguration + paths.toSeq.flatMap { path => + val input = HadoopInputFile.fromPath(new Path(path.stripPrefix("file:")), conf) + val reader = ParquetFileReader.open(input) + try { + reader.getFooter.getBlocks.asScala.flatMap { block => + block.getColumns.asScala + .filter(_.getPath.toDotString == columnName) + .map { column => + val bloom = reader.readBloomFilter(column) + assert(bloom != null, s"expected Bloom filter for $columnName in $path") + val out = new ByteArrayOutputStream(bloom.getBitsetSize) + bloom.writeTo(out) + out.toByteArray + } + } + } finally { + reader.close() + } + } + } + + private def parquetBloomFilterPresent(tableName: String, columnName: String): Boolean = { + val paths = spark + .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files ORDER BY file_path") + .collect() + .map(_.getString(0)) + val conf = spark.sparkContext.hadoopConfiguration + val present = paths.toSeq.flatMap { path => + val input = HadoopInputFile.fromPath(new Path(path.stripPrefix("file:")), conf) + val reader = ParquetFileReader.open(input) + try { + reader.getFooter.getBlocks.asScala.flatMap { block => + block.getColumns.asScala + .filter(_.getPath.toDotString == columnName) + .map(column => reader.readBloomFilter(column) != null) + } + } finally { + reader.close() + } + } + assert(present.nonEmpty, s"expected Parquet column $columnName in $tableName") + present.forall(identity) + } + + private def assertParquetBloomContainsInts( + tableName: String, + columnName: String, + values: Seq[Int]): Unit = { + val paths = spark + .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.data_files") + .collect() + .map(_.getString(0)) + val conf = spark.sparkContext.hadoopConfiguration + paths.foreach { path => + val input = HadoopInputFile.fromPath(new Path(path.stripPrefix("file:")), conf) + val reader = ParquetFileReader.open(input) + try { + reader.getFooter.getBlocks.asScala.foreach { block => + block.getColumns.asScala + .filter(_.getPath.toDotString == columnName) + .foreach { column => + val bloom = reader.readBloomFilter(column) + assert(bloom != null, s"expected Bloom filter for $columnName in $path") + values.foreach { value => + assert( + bloom.findHash(bloom.hash(value)), + s"Bloom filter false negative for $columnName=$value in $path") + } + } + } + } finally { + reader.close() + } + } + } + /** Native acceleration shared assumption -- currently just the Iceberg-on-classpath check. */ private def assumeNativeAcceleration(): Unit = { assume(icebergAvailable, "Iceberg not available in classpath") } + private def assumeIcebergBloomShapeProperties(): Unit = { + assume( + icebergSupportsBloomFpp && icebergSupportsBloomNdv, + "Iceberg runtime does not interpret per-column Bloom FPP/NDV properties") + } + + private def icebergSupportsBloomFpp: Boolean = + org.apache.comet.iceberg.IcebergReflection + .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX") + .isDefined + + private def icebergSupportsBloomNdv: Boolean = + org.apache.comet.iceberg.IcebergReflection + .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX") + .isDefined + /** * Every row-consuming operator must receive row-based input. Spark guarantees that by inserting * `ColumnarToRow` transitions in `ApplyColumnarRulesAndInsertTransitions`; an operator that diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala index 7c5631640c8..f31819303ca 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala @@ -261,28 +261,190 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes } } - test("fall-back: write.parquet.bloom-filter-max-bytes set") { + test("bloom-filter max-bytes accepts only representable power-of-two values") { + withDetectionCatalog { dir => + Seq("32", "524288", "1048576", "134217728").zipWithIndex.foreach { case (value, index) => + val table = s"bloom_max_ok_$index" + createTable( + dir, + table, + partitionSpec = "", + properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$value'")) + assertSupportLevelIs[Compatible](table) + } + + Seq("31", "33", "100", "134217729", "0", "-1", " 32", "garbage", "2147483648").zipWithIndex + .foreach { case (value, index) => + val table = s"bloom_max_bad_$index" + createTable( + dir, + table, + partitionSpec = "", + properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$value'")) + assertUnsupportedContainsAllowingWriteFailure( + table, + "write.parquet.bloom-filter-max-bytes") + } + } + } + + test("bloom-filter max-bytes=32 falls back only when parquet-mr ignores the cap") { + withDetectionCatalog { dir => + val minimumBytes = 32 + val naturallyMinimumNdv = 1 + val bindingFpp = 0.0001 + val bindingNdv = 1000000 + createTable( + dir, + "without_ndv", + partitionSpec = "", + properties = Some(s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'")) + assertSupportLevelIs[Compatible]("without_ndv") + + createTable( + dir, + "natural_minimum", + partitionSpec = "", + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='true', " + + s"'write.parquet.bloom-filter-ndv.column.id'='$naturallyMinimumNdv', " + + s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'")) + assertSupportLevelIs[Compatible]("natural_minimum") + + createTable( + dir, + "ignored_minimum_cap", + partitionSpec = "", + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='true', " + + 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'")) + if (icebergSupportsBloomFpp && icebergSupportsBloomNdv) { + assertUnsupportedContainsAllowingWriteFailure( + "ignored_minimum_cap", + "write.parquet.bloom-filter-max-bytes") + } else { + assertSupportLevelIs[Compatible]("ignored_minimum_cap") + } + } + } + + test("fall-back: bloom-filter NDV values that overflow parquet-mr sizing arithmetic") { + withDetectionCatalog { dir => + val largestNonOverflowingNdv = Long.MaxValue / 8L + val cases = Seq( + (largestNonOverflowingNdv, true), + (largestNonOverflowingNdv + 1L, false), + (1L << 61, false), + (Long.MaxValue, false)) + + cases.zipWithIndex.foreach { case ((ndv, compatible), index) => + val table = s"bloom_ndv_overflow_$index" + createTable( + dir, + table, + partitionSpec = "", + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='true', " + + s"'write.parquet.bloom-filter-ndv.column.id'='$ndv'")) + if (compatible || !icebergSupportsBloomNdv) { + assertSupportLevelIs[Compatible](table) + } else { + assertUnsupportedContainsAllowingWriteFailure( + table, + "write.parquet.bloom-filter-ndv.column.id") + } + } + } + } + + test("bloom-filter enabled=false with NDV follows runtime support and validates NDV") { + withDetectionCatalog { dir => + val configuredNdv = 1000 + createTable( + dir, + "false_with_ndv", + partitionSpec = "", + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='false', " + + s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'")) + assertSupportLevelIs[Compatible]("false_with_ndv") + + createTable( + dir, + "false_with_bad_ndv", + partitionSpec = "", + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='false', " + + "'write.parquet.bloom-filter-ndv.column.id'='garbage'")) + if (icebergSupportsBloomNdv) { + assertUnsupportedContainsAllowingWriteFailure( + "false_with_bad_ndv", + "write.parquet.bloom-filter-ndv") + } else { + assertSupportLevelIs[Compatible]("false_with_bad_ndv") + } + } + } + + test("bloom-filter enabled=false still validates FPP") { withDetectionCatalog { dir => createTable( dir, - "bloom_max", + "false_with_bad_fpp", partitionSpec = "", - properties = Some("'write.parquet.bloom-filter-max-bytes'='524288'")) - assertUnsupportedContains("bloom_max", "write.parquet.bloom-filter-max-bytes") + properties = Some( + "'write.parquet.bloom-filter-enabled.column.id'='false', " + + "'write.parquet.bloom-filter-fpp.column.id'='garbage'")) + if (icebergSupportsBloomFpp) { + assertUnsupportedContainsAllowingWriteFailure( + "false_with_bad_fpp", + "write.parquet.bloom-filter-fpp") + } else { + assertSupportLevelIs[Compatible]("false_with_bad_fpp") + } } } - test("fall-back: per-column bloom filter enabled") { + test("fall-back: invalid enabled-column FPP and NDV") { + withDetectionCatalog { dir => + Seq( + ("'write.parquet.bloom-filter-fpp.column.id'='0'", icebergSupportsBloomFpp), + ("'write.parquet.bloom-filter-fpp.column.id'='1'", icebergSupportsBloomFpp), + ("'write.parquet.bloom-filter-fpp.column.id'='NaN'", icebergSupportsBloomFpp), + // Positive and below one, but too small for any integer NDV to encode the requested + // power-of-two allocation through parquet-rs's NDV/FPP API. + ("'write.parquet.bloom-filter-fpp.column.id'='4.9E-324'", icebergSupportsBloomFpp), + ("'write.parquet.bloom-filter-ndv.column.id'='0'", icebergSupportsBloomNdv), + ( + "'write.parquet.bloom-filter-ndv.column.id'='garbage'", + icebergSupportsBloomNdv)).zipWithIndex + .foreach { case ((property, interpreted), index) => + val table = s"bloom_shape_bad_$index" + createTable( + dir, + table, + partitionSpec = "", + properties = + Some(s"'write.parquet.bloom-filter-enabled.column.id'='true', $property")) + if (interpreted) { + assertUnsupportedContainsAllowingWriteFailure(table, "write.parquet.bloom-filter") + } else { + assertSupportLevelIs[Compatible](table) + } + } + } + } + + test("Compatible when a per-column bloom filter is enabled") { withDetectionCatalog { dir => createTable( dir, "bloom_col", partitionSpec = "", properties = Some("'write.parquet.bloom-filter-enabled.column.id'='true'")) - assertUnsupportedContains( - "bloom_col", - "write.parquet.bloom-filter-enabled.column.id", - "true") + assertSupportLevelIs[Compatible]("bloom_col") } } @@ -890,6 +1052,16 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes } } + private def icebergSupportsBloomFpp: Boolean = + IcebergReflection + .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX") + .isDefined + + private def icebergSupportsBloomNdv: Boolean = + IcebergReflection + .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX") + .isDefined + /** * Runs Spark's transition insertion followed by [[EliminateRedundantTransitions]] over a * hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source` plan and returns the diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala index f1dd09ccf7e..03e56eb81a0 100644 --- a/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala +++ b/spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala @@ -176,6 +176,79 @@ class IcebergWriteProtoTranslationSuite extends AnyFunSuite { assert(settings.getPageRowLimit == 1000) } + test("per-column bloom filter properties are translated deterministically") { + val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix + val settings = buildParquetSettings( + Map( + s"${prefix}region" -> "TRUE", + s"${prefix}id" -> "true", + s"${prefix}amount" -> "false", + s"${Keys.ParquetBloomFilterColumnFppPrefix}region" -> "0.02", + s"${Keys.ParquetBloomFilterColumnNdvPrefix}id" -> "1234"), + TestCreatedBy) + assert(settings.getBloomFilterEnabledColumnsList == java.util.Arrays.asList("id", "region")) + assert(settings.getBloomFilterMaxBytes == 1024L * 1024L) + assert(settings.getBloomFilterFppByColumnMap.get("id") == 0.01d) + assert(settings.getBloomFilterFppByColumnMap.get("region") == 0.02d) + assert(settings.getBloomFilterNdvByColumnMap.get("id") == 1234L) + assert(!settings.getBloomFilterNdvByColumnMap.containsKey("region")) + assert(!settings.getBloomFilterFppByColumnMap.containsKey("amount")) + } + + test("an explicit NDV re-enables a bloom filter after enabled=false") { + val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix + val configuredFpp = 0.02d + val configuredNdv = 1234L + val settings = buildParquetSettings( + Map( + s"${prefix}id" -> "false", + s"${Keys.ParquetBloomFilterColumnFppPrefix}id" -> configuredFpp.toString, + s"${Keys.ParquetBloomFilterColumnNdvPrefix}id" -> configuredNdv.toString), + TestCreatedBy) + + assert(settings.getBloomFilterEnabledColumnsList == java.util.Arrays.asList("id")) + assert(settings.getBloomFilterFppByColumnMap.get("id") == configuredFpp) + assert(settings.getBloomFilterNdvByColumnMap.get("id") == configuredNdv) + } + + test("bloom filter properties use physical Parquet paths for list and map leaves") { + val prefix = Keys.ParquetBloomFilterColumnEnabledPrefix + val settings = buildParquetSettings( + Map( + s"${prefix}tags.element" -> "true", + s"${prefix}attrs.key" -> "true", + s"${prefix}attrs.value" -> "true", + s"${prefix}missing" -> "true", + s"${Keys.ParquetBloomFilterColumnFppPrefix}tags.element" -> "0.02", + s"${Keys.ParquetBloomFilterColumnNdvPrefix}attrs.value" -> "1234"), + TestCreatedBy, + Map( + "tags.element" -> "tags.list.element", + "attrs.key" -> "attrs.key_value.key", + "attrs.value" -> "attrs.key_value.value")) + + assert( + settings.getBloomFilterEnabledColumnsList == java.util.Arrays + .asList("attrs.key_value.key", "attrs.key_value.value", "tags.list.element")) + assert(settings.getBloomFilterFppByColumnMap.get("tags.list.element") == 0.02d) + assert(settings.getBloomFilterNdvByColumnMap.get("attrs.key_value.value") == 1234L) + assert(!settings.getBloomFilterEnabledColumnsList.contains("missing")) + } + + test("Iceberg bloom filter defaults and explicit max are translated exactly") { + val enabled = Keys.ParquetBloomFilterColumnEnabledPrefix + "id" + val defaults = buildParquetSettings(Map(enabled -> "true"), TestCreatedBy) + assert(defaults.getBloomFilterMaxBytes == Defaults.BloomFilterMaxBytes) + assert(Defaults.BloomFilterMaxBytes == 1024 * 1024) + assert(defaults.getBloomFilterFppByColumnMap.get("id") == Defaults.BloomFilterFpp) + assert(Defaults.BloomFilterFpp == 0.01d) + + val explicit = buildParquetSettings( + Map(enabled -> "true", Keys.ParquetBloomFilterMaxBytes -> "67108864"), + TestCreatedBy) + assert(explicit.getBloomFilterMaxBytes == 64L * 1024L * 1024L) + } + test("size properties are parsed with Java Integer.parseInt semantics") { // No trimming and no values past Int.MaxValue -- exactly what iceberg-java's // PropertyUtil.propertyAsInt would do. The eligibility gate declines these values