Merge pull request JanKaul#379 from cedricziel/feat/parquet-bloom-column-properties

feat: honor per-column bloom-filter table properties in the Parquet writer

Author: JanKaul

iceberg-rust-spec/src/spec/table_metadata.rs

@@ -49,6 +49,10 @@ pub const WRITE_OBJECT_STORAGE_ENABLED: &str = "write.object-storage.enabled";
 pub const WRITE_DATA_PATH: &str = "write.data.path";
 pub const WRITE_METADATA_METRICS_DISTINCT_COUNTS_ENABLED: &str =
     "write.metadata.metrics.distinct-counts.enabled";
+/// Per-column Parquet bloom-filter toggle: append the column name, e.g.
+/// `write.parquet.bloom-filter-enabled.column.label_env = "true"`.
+pub const WRITE_PARQUET_BLOOM_FILTER_ENABLED_COLUMN_PREFIX: &str =
+    "write.parquet.bloom-filter-enabled.column.";
 
 pub use _serde::{TableMetadataV1, TableMetadataV2, TableMetadataV3};
 

iceberg-rust/src/arrow/write.rs

@@ -51,7 +51,7 @@ use iceberg_rust_spec::{
     spec::{manifest::DataFile, schema::Schema, values::Value},
     table_metadata::{
         self, WRITE_DATA_PATH, WRITE_METADATA_METRICS_DISTINCT_COUNTS_ENABLED,
-        WRITE_OBJECT_STORAGE_ENABLED,
+        WRITE_OBJECT_STORAGE_ENABLED, WRITE_PARQUET_BLOOM_FILTER_ENABLED_COLUMN_PREFIX,
     },
     util::strip_prefix,
 };
@@ -62,6 +62,7 @@ use parquet::{
         metadata::{KeyValue, ParquetMetaData},
         properties::WriterProperties,
     },
+    schema::types::ColumnPath,
 };
 use uuid::Uuid;
 
@@ -514,6 +515,7 @@ async fn create_arrow_writer(
 
     let mut props_builder =
         WriterProperties::builder().set_compression(Compression::ZSTD(ZstdLevel::try_new(1)?));
+    props_builder = apply_bloom_filter_properties(props_builder, table_properties);
     if estimate_distinct_count {
         props_builder = props_builder.set_key_value_metadata(Some(vec![KeyValue::new(
             ICEBERG_ESTIMATE_INT64_DISTINCT_COUNT_META_KEY.to_owned(),
@@ -531,6 +533,29 @@ async fn create_arrow_writer(
     ))
 }
 
+/// Applies per-column bloom-filter table properties to the writer builder.
+///
+/// Honors the standard Iceberg property
+/// `write.parquet.bloom-filter-enabled.column.<name>` = `true`/`false`
+/// per column. Unrelated properties are ignored.
+fn apply_bloom_filter_properties(
+    mut props_builder: parquet::file::properties::WriterPropertiesBuilder,
+    table_properties: &HashMap<String, String>,
+) -> parquet::file::properties::WriterPropertiesBuilder {
+    for (key, value) in table_properties {
+        if let Some(column) = key.strip_prefix(WRITE_PARQUET_BLOOM_FILTER_ENABLED_COLUMN_PREFIX) {
+            let enabled = value.eq_ignore_ascii_case("true");
+            // Parquet addresses nested columns by path parts; a dotted name
+            // passed as one string would be treated as a single segment and
+            // silently never match.
+            let path_parts: Vec<String> = column.split('.').map(String::from).collect();
+            props_builder = props_builder
+                .set_column_bloom_filter_enabled(ColumnPath::from(path_parts), enabled);
+        }
+    }
+    props_builder
+}
+
 /// Generates a unique file path for a Parquet data file.
 ///
 /// This function creates a unique file path by combining the data location, partition path,
@@ -614,6 +639,73 @@ fn record_batch_size(batch: &RecordBatch) -> usize {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
+
+    #[test]
+    fn bloom_filter_properties_apply_per_column() {
+        let table_properties = HashMap::from([
+            (
+                "write.parquet.bloom-filter-enabled.column.label_env".to_string(),
+                "true".to_string(),
+            ),
+            (
+                "write.parquet.bloom-filter-enabled.column.body".to_string(),
+                "false".to_string(),
+            ),
+            ("write.data.path".to_string(), "s3://x".to_string()),
+        ]);
+        let props =
+            apply_bloom_filter_properties(WriterProperties::builder(), &table_properties).build();
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from("label_env"))
+            .is_some());
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from("body"))
+            .is_none());
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from("other"))
+            .is_none());
+    }
+
+    #[test]
+    fn bloom_filter_properties_apply_to_nested_columns() {
+        let table_properties = HashMap::from([(
+            "write.parquet.bloom-filter-enabled.column.my_struct.label_env".to_string(),
+            "true".to_string(),
+        )]);
+        let props =
+            apply_bloom_filter_properties(WriterProperties::builder(), &table_properties).build();
+        // Parquet addresses nested columns by path parts, not by the dotted string.
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from(vec![
+                "my_struct".to_string(),
+                "label_env".to_string()
+            ]))
+            .is_some());
+    }
+
+    #[test]
+    fn bloom_filter_property_values_are_case_insensitive() {
+        let table_properties = HashMap::from([
+            (
+                "write.parquet.bloom-filter-enabled.column.label_env".to_string(),
+                "TRUE".to_string(),
+            ),
+            (
+                "write.parquet.bloom-filter-enabled.column.body".to_string(),
+                "False".to_string(),
+            ),
+        ]);
+        let props =
+            apply_bloom_filter_properties(WriterProperties::builder(), &table_properties).build();
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from("label_env"))
+            .is_some());
+        assert!(props
+            .bloom_filter_properties(&ColumnPath::from("body"))
+            .is_none());
+    }
+
     use iceberg_rust_spec::{
         partition::BoundPartitionField,
         types::{StructField, Type},