apache / apache/datafusion

Bloom filters and statistics not being used for Map keys/values

Open
#17,221 0 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Describe the bug

Continuing from the thread at https://discord.com/channels/885562378132000778/1402989687324414046
We are trying to add bloom filters to values of a Map field and while we do it see it available in the parquet file, we don't see it getting used while querying

### To Reproduce

```rust
use arrow::array::{ArrayRef, MapArray, StringArray, StructArray};
use arrow::buffer::OffsetBuffer;
use arrow::datatypes::{DataType, Field, Fields, Schema};
use arrow::record_batch::RecordBatch;
use chrono::{DateTime, Duration, Utc};
use datafusion::prelude::*;
use parquet::arrow::arrow_writer::ArrowWriter;
use parquet::basic::Compression;
use parquet::file::properties::{EnabledStatistics, WriterProperties};
use std::collections::HashMap;
use std::fs::File;
use std::sync::Arc;
use tempfile::TempDir;

#[tokio::main]
async fn main() -> Result<(), Box> {
println!("Creating parquet file with bloom filters...");

let temp_dir = TempDir::new()?;
let parquet_path = temp_dir.path().join("bloom_filter_data.parquet");

let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new_map(
"SpanAttributesString",
"key_value",
Arc::new(Field::new("key", DataType::Utf8, false)),
Arc::new(Field::new("value", DataType::Utf8, true)),
false, // keys are not sorted
true, // map itself is nullable
),
]));

/*
Sample data:
+----+------------------------------------+
| id | SpanAttributesString |
+----+------------------------------------+
| 1 | {key_1: 2025-01-01T00:00:01+00:00} |
| 2 | {key_1: 2025-01-01T00:00:02+00:00} |
| 3 | {key_1: 2025-01-01T00:00:03+00:00} |
| 4 | {key_1: 2025-01-01T00:00:04+00:00} |
| 5 | {key_1: 2025-01-01T00:00:05+00:00} |
| 6 | {key_1: 2025-01-01T00:00:06+00:00} |
| 7 | {key_1: 2025-01-01T00:00:07+00:00} |
| 8 | {key_1: 2025-01-01T00:00:08+00:00} |
| 9 | {key_1: 2025-01-01T00:00:09+00:00} |
| 10 | {key_1: 2025-01-01T00:00:10+00:00} |
+----+------------------------------------+
*/

let id_col_path_value = parquet::schema::types::ColumnPath::from(vec!["id".to_string()]);
let map_col_path_value = parquet::schema::types::ColumnPath::from(vec![
"SpanAttributesString".to_string(),
"key_value".to_string(),
"value".to_string(),
]);

let writer_properties = WriterProperties::builder()
.set_compression(Compression::LZ4_RAW)
.set_statistics_enabled(EnabledStatistics::Chunk)
.set_column_bloom_filter_enabled(id_col_path_value.clone(), true)
.set_column_bloom_filter_ndv(id_col_path_value.clone(), 1000)
.set_column_bloom_filter_fpp(id_col_path_value, 0.00001)
.set_column_bloom_filter_enabled(map_col_path_value.clone(), true)
.set_column_bloom_filter_ndv(map_col_path_value.clone(), 1000)
.set_column_bloom_filter_fpp(map_col_path_value, 0.00001)
.build();

let file = File::create(&parquet_path)?;

let mut writer = ArrowWriter::try_new(file, schema.clone(), Some(writer_properties))?;

// Create sample data with repetitive patterns to test bloom filter efficiency
let mut ids = Vec::::new();
let mut maps = Vec::new();

let dt = DateTime::parse_from_rfc3339("2025-01-01T00:00:00Z")?.with_timezone(&Utc);

// Generate data with specific patterns
for i in 1..=10000 {
ids.push(i.to_string());
maps.push(HashMap::from([(
"key_1".to_string(),
(dt + Duration::seconds(i as i64)).to_rfc3339().to_string(),
)]));

if i % 1000 == 0 {
println!("Writing batch {}", i);
let ids_array = Arc::new(StringArray::from(ids.clone()));
let mut keys = Vec::new();
let mut values = Vec::new();
let mut lens = Vec::new();

for map in maps.clone() {
keys.extend(map.keys().cloned().collect::>());
values.extend(map.values().cloned().collect::>());
lens.push(map.len());
}

let keys_array = Arc::new(StringArray::from(keys));
let values_array = Arc::new(StringArray::from(values));
let entries = StructArray::from(vec![
(
Arc::new(Field::new("key", DataType::Utf8, false)),
keys_array as ArrayRef,
),
(
Arc::new(Field::new("value", DataType::Utf8, true)),
values_array as ArrayRef,
),
]);

let field = Arc::new(Field::new(
"key_value",
DataType::Struct(Fields::from(vec![
Field::new("key", DataType::Utf8, false),
Field::new("value", DataType::Utf8, true),
])),
false,
));

let map_column = Arc::new(MapArray::try_new(
field,
OffsetBuffer::from_lengths(lens),
entries,
None,
false,
)?);

let record_batch = RecordBatch::try_new(schema.clone(), vec![ids_array, map_column])?;

writer.write(&record_batch)?;
writer.flush()?;

ids.clear();
maps.clear();
}
}

writer.close()?;

println!("Parquet file created at: {parquet_path:?}");
println!("Bloom filters enabled for columns");

let ctx = SessionContext::new_with_config(SessionConfig::from_env()?);
ctx.register_parquet(
"test_data",
parquet_path.to_str().unwrap(),
ParquetReadOptions::default(),
)
.await?;

println!("\nSample data:");
let df = ctx
.sql("SELECT \"id\", \"SpanAttributesString\" FROM test_data LIMIT 10")
.await?;
df.show().await?;

println!("\n=== First Query: Test bloom filter for id ===");
let query1 = "
EXPLAIN ANALYZE
SELECT *
FROM test_data
WHERE \"id\" = '1345'
";
let result1 = ctx.sql(query1).await?;
result1.show().await?;

println!("\n=== Second Query: Test bloom filter for map ===");
let query2 = "
EXPLAIN ANALYZE
SELECT *
FROM test_data
WHERE
array_contains(map_values(\"SpanAttributesString\"), '2025-01-01T00:00:05+00:00')
";
let result2 = ctx.sql(query2).await?;
result2.show().await?;

println!("\n=== Third Query: Test bloom filter for map ===");
let query3 = "
EXPLAIN ANALYZE
SELECT *
FROM test_data
WHERE
array_has_any(map_values(\"SpanAttributesString\"), ['2025-01-01T00:00:05+00:00', '2025-01-01T00:30:56+00:00'])
";

let result3 = ctx.sql(query3).await?;
result3.show().await?;

Ok(())
}
```

```
Creating parquet file with bloom filters...
Writing batch 1000
Writing batch 2000
Writing batch 3000
Writing batch 4000
Writing batch 5000
Writing batch 6000
Writing batch 7000
Writing batch 8000
Writing batch 9000
Writing batch 10000
Parquet file created at: "/var/folders/qw/rd2m1w2x4z7ffl2c25ys5gw80000gn/T/.tmp5DkioK/bloom_filter_data.parquet"
Bloom filters enabled for columns

Sample data:
+----+------------------------------------+
| id | SpanAttributesString |
+----+------------------------------------+
| 1 | {key_1: 2025-01-01T00:00:01+00:00} |
| 2 | {key_1: 2025-01-01T00:00:02+00:00} |
| 3 | {key_1: 2025-01-01T00:00:03+00:00} |
| 4 | {key_1: 2025-01-01T00:00:04+00:00} |
| 5 | {key_1: 2025-01-01T00:00:05+00:00} |
| 6 | {key_1: 2025-01-01T00:00:06+00:00} |
| 7 | {key_1: 2025-01-01T00:00:07+00:00} |
| 8 | {key_1: 2025-01-01T00:00:08+00:00} |
| 9 | {key_1: 2025-01-01T00:00:09+00:00} |
| 10 | {key_1: 2025-01-01T00:00:10+00:00} |
+----+------------------------------------+

=== First Query: Test bloom filter for id ===
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=1, elapsed_compute=61.248µs] |
| | FilterExec: id@0 = 1345, metrics=[output_rows=1, elapsed_compute=444.965µs] |
| | RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, metrics=[fetch_time=4.552625ms, repartition_time=1ns, send_time=18.424µs] |
| | DataSourceExec: file_groups={1 group: [[var/folders/qw/rd2m1w2x4z7ffl2c25ys5gw80000gn/T/.tmp5DkioK/bloom_filter_data.parquet]]}, projection=[id, SpanAttributesString], file_type=parquet, predicate=id@0 = 1345, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1345 AND 1345 <= id_max@1, required_guarantees=[id in (1345)] |
| | , metrics=[output_rows=1000, elapsed_compute=1ns, batches_splitted=0, bytes_scanned=24817, file_open_errors=0, file_scan_errors=0, files_ranges_pruned_statistics=0, num_predicate_creation_errors=0, page_index_rows_matched=1000, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=1, row_groups_matched_statistics=3, row_groups_pruned_bloom_filter=2, row_groups_pruned_statistics=7, bloom_filter_eval_time=655.376µs, metadata_load_time=984.585µs, page_index_eval_time=139.001µs, row_pushdown_eval_time=2ns, statistics_eval_time=199.876µs, time_elapsed_opening=3.237792ms, time_elapsed_processing=4.067038ms, time_elapsed_scanning_total=1.304626ms, time_elapsed_scanning_until_data=1.242ms] |
| | |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

=== Second Query: Test bloom filter for map ===
+-------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=1, elapsed_compute=21.127µs] |
| | FilterExec: array_has(map_values(SpanAttributesString@1), 2025-01-01T00:00:05+00:00), metrics=[output_rows=1, elapsed_compute=3.572ms] |
| | RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, metrics=[fetch_time=12.772459ms, repartition_time=1ns, send_time=41.833µs] |
| | DataSourceExec: file_groups={1 group: [[var/folders/qw/rd2m1w2x4z7ffl2c25ys5gw80000gn/T/.tmp5DkioK/bloom_filter_data.parquet]]}, projection=[id, SpanAttributesString], file_type=parquet, metrics=[output_rows=10000, elapsed_compute=1ns, batches_splitted=0, bytes_scanned=119161, file_open_errors=0, file_scan_errors=0, files_ranges_pruned_statistics=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=0, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=0, bloom_filter_eval_time=2ns, metadata_load_time=557.834µs, page_index_eval_time=2ns, row_pushdown_eval_time=2ns, statistics_eval_time=2ns, time_elapsed_opening=609µs, time_elapsed_processing=11.987543ms, time_elapsed_scanning_total=12.529081ms, time_elapsed_scanning_until_data=1.254375ms] |
| | |
+-------------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

=== Third Query: Test bloom filter for map ===
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=2, elapsed_compute=38.502µs] |
| | FilterExec: array_has_any(map_values(SpanAttributesString@1), [2025-01-01T00:00:05+00:00, 2025-01-01T00:30:56+00:00]), metrics=[output_rows=2, elapsed_compute=18.530457ms] |
| | RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, metrics=[fetch_time=13.543747ms, repartition_time=1ns, send_time=58.457µs] |
| | DataSourceExec: file_groups={1 group: [[var/folders/qw/rd2m1w2x4z7ffl2c25ys5gw80000gn/T/.tmp5DkioK/bloom_filter_data.parquet]]}, projection=[id, SpanAttributesString], file_type=parquet, metrics=[output_rows=10000, elapsed_compute=1ns, batches_splitted=0, bytes_scanned=119161, file_open_errors=0, file_scan_errors=0, files_ranges_pruned_statistics=0, num_predicate_creation_errors=0, page_index_rows_matched=0, page_index_rows_pruned=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_matched_bloom_filter=0, row_groups_matched_statistics=0, row_groups_pruned_bloom_filter=0, row_groups_pruned_statistics=0, bloom_filter_eval_time=2ns, metadata_load_time=550.835µs, page_index_eval_time=2ns, row_pushdown_eval_time=2ns, statistics_eval_time=2ns, time_elapsed_opening=609.875µs, time_elapsed_processing=12.603708ms, time_elapsed_scanning_total=14.789501ms, time_elapsed_scanning_until_data=1.163459ms] |
| | |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```

### Expected behavior

Statistics and bloom filters should be used to prune row groups

### Additional context

_No response_

Contributor guide

Open the contributing guide

Research direction

Reproduce the issue with the Rust program using ArrowWriter, SessionContext, register_parquet, and the three EXPLAIN ANALYZE queries. Start by comparing the row-group behavior for the scalar id predicate with the map_values predicates; done means statistics and bloom filters on Map keys or values are demonstrably used during querying without changing query results.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, sql
Domain
databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.