You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "hengfeiyang (via GitHub)" <gi...@apache.org> on 2023/10/14 01:16:29 UTC
[PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
hengfeiyang opened a new pull request, #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821
## Which issue does this PR close?
Impl #4512
## Rationale for this change
## What changes are included in this PR?
I implement `bloom filter` support when we pruning `row group`.
I already created a PR and merged for support read `bloom filter` with `async` https://github.com/apache/arrow-rs/pull/4917
But before `arrow-rs` team make a new release we need to test it with a special `commit id` in Cargo.toml
Let's use this to review this PR, I will switch to the new version after `arrow-rs` team made the new release.
## Are these changes tested?
- [ ] Test with a parquet file that can pruning `row group` by `bloom filter`
- [ ] Test with a parquet file that can't pruning `row group` by `bloom filter`
- [ ] Test with a parquet without `bloom filter`
## Are there any user-facing changes?
no
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1762461298
I already created a PR and merged for support read `bloom filter` with `async` https://github.com/apache/arrow-rs/pull/4917
But before `arrow-rs` team make a new release we need to test it with a special `commit id` in Cargo.toml
Let's use this to discus this PR, I will switch to the new version after `arrow-rs` team made the new release.
@Ted-Jiang @alamb Can you review this PR, any suggestions? Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "Ted-Jiang (via GitHub)" <gi...@apache.org>.
Ted-Jiang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366398648
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
IMO @hengfeiyang call `orig_expr ` will not get the `Inlist` rewrite here, maybe you should write a test for `select * from table col1 in (xxx)` π€
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
IMO @hengfeiyang call `orig_expr ` will not get the `Inlist` rewrite here, maybe you should write a test for `select * from table col1 in (xxx)` π€
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1365875743
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
Review Comment:
You could potentially make this slightly more concise like this:
```suggestion
let Some(expr) = expr else {
// unsupported predicate
return false;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "Ted-Jiang (via GitHub)" <gi...@apache.org>.
Ted-Jiang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366383819
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
IMO, `Inlist` is also one commonly used case, we should support later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "Ted-Jiang (via GitHub)" <gi...@apache.org>.
Ted-Jiang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366384382
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Maybe we can this support later π
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Maybe we can this support later π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1779708060
>
My understanding is that bloom filter returns false positives which you have to filter exactly (without a bloom filter) in a second step. I was referring to the second step. I never wrote such algorithm before so there is a good chance I donβt know what Iβm talking about π
Hi @alippai -- what this PR does is use the (optional) Bloom Filter that is stored in parquet files to evaluate equality predicates more efficiently -- for example, `a = 4` type predicates.
In general it would be interesting to contemplate using a bloom filter for COUNT(distinct ...) type predicates, though the existing hash table based implementation is pretty fast
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1762466012
To test this feature we need to enable an option:
`datafusion.execution.parquet.bloom_filter_enabled` default is `false`
Or just set an ENV:
```
DATAFUSION_EXECUTION_PARQUET_BLOOM_FILTER_ENABLED=true
```
This is my test use the `datafusion-cli` in this branch:
```
DataFusion CLI v32.0.0
β― create external table tbl stored as parquet location '/Users/yanghengfei/code/rust/github.com/zinclabs/openobserve/data/bloomfilter/stream/files/default/logs/traces/';
0 rows in set. Query took 0.042 seconds.
β― select count(*) from tbl;
+----------+
| COUNT(*) |
+----------+
| 60992352 |
+----------+
1 row in set. Query took 0.083 seconds.
β― SELECT * FROM tbl where (_timestamp >= 1694908802204000 AND _timestamp < 1696593531571122) AND trace_id='3c7dbf90d1a66e3faffa344519c3bac0' LIMIT 150;
+------------------+----------+---------------------+--------+-------+...
| _timestamp | duration | end_time | events | flags | trace_id |...
+------------------+----------+---------------------+--------+-------+...
| 1694984600199046 | 2008886 | 1694984602207932061 | [] | 1 | ...
| 1694984600199000 | 2009958 | 1694984602208958875 | [] | 1 | ...
+------------------+----------+---------------------+--------+-------+...
2 rows in set. Query took 4.086 seconds.
β― set datafusion.execution.parquet.bloom_filter_enabled=true;
0 rows in set. Query took 0.000 seconds.
β― SELECT * FROM tbl where (_timestamp >= 1694908802204000 AND _timestamp < 1696593531571122) AND trace_id='3c7dbf90d1a66e3faffa344519c3bac0' LIMIT 150;
+------------------+----------+---------------------+--------+-------+...
| _timestamp | duration | end_time | events | flags | trace_id |...
+------------------+----------+---------------------+--------+-------+...
| 1694984600199046 | 2008886 | 1694984602207932061 | [] | 1 | ...
| 1694984600199000 | 2009958 | 1694984602208958875 | [] | 1 | ...
+------------------+----------+---------------------+--------+-------+...
2 rows in set. Query took 0.234 seconds.
```
The information of this directory:
```shell
$ cd /Users/yanghengfei/code/rust/github.com/zinclabs/openobserve/data/bloomfilter/stream/files/default/logs/traces/
du -sh .
2.8G .
$ find . -name "*.parquet"|wc -l
326
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1767211985
> But before `arrow-rs` team makes a new release we need to test it with a special `commit id` in `Cargo.toml`
Thank you -- I expect that we'll have a new arrow-rs release candidate later this week and the next release in about 7 days from now, for your scheduling planning
> Let's use this to discuss this PR, I will switch to the new version after `arrow-rs` team makes the new release.
>
> @Ted-Jiang @alamb Can you review this PR, any suggestions? Thanks
Thank you @hengfeiyang -- I plan to review this PR carefully later today or tomorrow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366652329
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
+ match expr.op() {
+ Operator::And => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left || right
+ }
+ Operator::Or => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left && right
+ }
+ Operator::Eq => {
+ if let Some((col, val)) = Self::check_expr_is_col_equal_const(expr) {
+ if let Some(sbbf) = column_sbbf.get(col.name()) {
+ match val {
+ ScalarValue::Utf8(Some(v)) => !sbbf.check(&v.as_str()),
+ ScalarValue::Boolean(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int16(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int8(Some(v)) => !sbbf.check(&v),
+ _ => false,
+ }
+ } else {
+ false
+ }
+ } else {
+ false
+ }
+ }
+ _ => false,
+ }
+ }
+
+ fn get_predicate_columns(
+ expr: Option<&phys_expr::BinaryExpr>,
+ ) -> Option<HashSet<String>> {
+ match expr {
Review Comment:
Merged
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
+ match expr.op() {
+ Operator::And => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left || right
+ }
+ Operator::Or => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left && right
+ }
+ Operator::Eq => {
+ if let Some((col, val)) = Self::check_expr_is_col_equal_const(expr) {
+ if let Some(sbbf) = column_sbbf.get(col.name()) {
+ match val {
+ ScalarValue::Utf8(Some(v)) => !sbbf.check(&v.as_str()),
+ ScalarValue::Boolean(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int16(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int8(Some(v)) => !sbbf.check(&v),
+ _ => false,
+ }
+ } else {
+ false
+ }
+ } else {
+ false
+ }
+ }
+ _ => false,
+ }
+ }
+
+ fn get_predicate_columns(
+ expr: Option<&phys_expr::BinaryExpr>,
+ ) -> Option<HashSet<String>> {
+ match expr {
Review Comment:
Merged, thank you very much.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366680038
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -84,6 +84,9 @@ pub struct ParquetExec {
/// Override for `Self::with_enable_page_index`. If None, uses
/// values from base_config
enable_page_index: Option<bool>,
+ /// Override for `Self::with_enable_bloom_filter`. If None, uses
+ /// values from base_config
+ enable_bloom_filter: Option<bool>,
Review Comment:
so, i can use `set datafusion.execution.parquet.bloom_filter_enabled=true;` in `datafusion-cli` to enable / disable it in a session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alippai (via GitHub)" <gi...@apache.org>.
alippai commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1778249260
My understanding is that bloom filter returns false positives which you have to filter exactly (without a bloom filter) in a second step. I was referring to the second step. I never wrote such algorithm before so there is a good chance I donβt know what Iβm talking about π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1771442506
I also have an idea of how to make this code potentially even more general using interval analysis that I am trying out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366637327
##########
datafusion/sqllogictest/test_files/json.slt:
##########
@@ -58,8 +58,10 @@ AggregateExec: mode=Final, gby=[], aggr=[COUNT(*)]
------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
--------JsonExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/2.json]]}, projection=[a]
-query error DataFusion error: Schema error: No field named mycol\.
+query N
Review Comment:
Yep, i will rollback in this PR, and then should fix in another PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366442576
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
Sure, i will add a test for `col in(xxx)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1776239728
@alamb Thanks, already rebased the main branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alippai (via GitHub)" <gi...@apache.org>.
alippai commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1776507315
My question might be naive, but if count(distinct values) == number of rows then does this MR use this info (eg returning early from the scan because we know itβs exactly one row as the column is a primary key)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366643868
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -846,4 +1120,185 @@ mod tests {
let execution_props = ExecutionProps::new();
create_physical_expr(expr, &df_schema, schema, &execution_props).unwrap()
}
+
+ #[tokio::test]
+ async fn test_row_group_bloom_filter_pruning_predicate_simple_expr() {
+ // load parquet file
+ let testdata = datafusion_common::test_util::parquet_test_data();
+ let file_name = "data_index_bloom_encoding_stats.parquet";
+ let path = format!("{testdata}/{file_name}");
+ let data = bytes::Bytes::from(std::fs::read(path).unwrap());
+
+ // generate pruning predicate
+ let schema = Schema::new(vec![Field::new("String", DataType::Utf8, false)]);
+ let expr = col(r#""String""#).eq(lit("Hello_Not_Exists"));
+ let expr = logical2physical(&expr, &schema);
+ let pruning_predicate =
+ PruningPredicate::try_new(expr, Arc::new(schema)).unwrap();
+
+ let row_groups = vec![0];
+ let pruned_row_groups = test_row_group_bloom_filter_pruning_predicate(
+ file_name,
+ data,
+ &pruning_predicate,
+ &row_groups,
+ )
+ .await
+ .unwrap();
+ assert!(pruned_row_groups.is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_row_group_bloom_filter_pruning_predicate_partial_expr() {
+ // load parquet file
+ let testdata = datafusion_common::test_util::parquet_test_data();
+ let file_name = "data_index_bloom_encoding_stats.parquet";
+ let path = format!("{testdata}/{file_name}");
+ let data = bytes::Bytes::from(std::fs::read(path).unwrap());
+
+ // generate pruning predicate
+ let schema = Schema::new(vec![Field::new("String", DataType::Utf8, false)]);
+ let expr = col(r#""String""#)
Review Comment:
I deleted `test_row_group_bloom_filter_pruning_predicate_partial_expr`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1365916423
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
Review Comment:
I think this is copy/paste from PruningPredicate -- this expression isn't rewritten
```suggestion
/// Actual pruning predicate
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366392718
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Yes, i found that. will add a comment for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366678958
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -84,6 +84,9 @@ pub struct ParquetExec {
/// Override for `Self::with_enable_page_index`. If None, uses
/// values from base_config
enable_page_index: Option<bool>,
+ /// Override for `Self::with_enable_bloom_filter`. If None, uses
+ /// values from base_config
+ enable_bloom_filter: Option<bool>,
Review Comment:
I saw that already have the config https://github.com/apache/arrow-datafusion/blob/c97a0d2ebb8697855edd5b56904accbfa9a80290/datafusion/common/src/config.rs#L368C13-L368C33
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "Ted-Jiang (via GitHub)" <gi...@apache.org>.
Ted-Jiang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366399866
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
BTW, this is a awesome feature, thx for your great work! π
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
BTW, this is a awesome feature, thx for your great work! π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366442576
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
Thank you, and i will add a test for `col in(xxx)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366808645
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -84,6 +84,9 @@ pub struct ParquetExec {
/// Override for `Self::with_enable_page_index`. If None, uses
/// values from base_config
enable_page_index: Option<bool>,
+ /// Override for `Self::with_enable_bloom_filter`. If None, uses
+ /// values from base_config
+ enable_bloom_filter: Option<bool>,
Review Comment:
Looks good to me. Thank you
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -246,6 +250,18 @@ impl ParquetExec {
.unwrap_or(config_options.execution.parquet.enable_page_index)
}
+ /// If enabled, the reader will read by the bloom filter
+ pub fn with_enable_bloom_filter(mut self, enable_bloom_filter: bool) -> Self {
+ self.enable_bloom_filter = Some(enable_bloom_filter);
+ self
+ }
+
+ /// Return the value described in [`Self::with_enable_bloom_filter`]
+ fn enable_bloom_filter(&self, config_options: &ConfigOptions) -> bool {
+ self.enable_bloom_filter
+ .unwrap_or(config_options.execution.parquet.bloom_filter_enabled)
Review Comment:
π
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -246,6 +250,18 @@ impl ParquetExec {
.unwrap_or(config_options.execution.parquet.enable_page_index)
}
+ /// If enabled, the reader will read by the bloom filter
+ pub fn with_enable_bloom_filter(mut self, enable_bloom_filter: bool) -> Self {
+ self.enable_bloom_filter = Some(enable_bloom_filter);
+ self
+ }
+
+ /// Return the value described in [`Self::with_enable_bloom_filter`]
+ fn enable_bloom_filter(&self, config_options: &ConfigOptions) -> bool {
+ self.enable_bloom_filter
+ .unwrap_or(config_options.execution.parquet.bloom_filter_enabled)
Review Comment:
π
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
Review Comment:
BTW I have some ideas (working on them in https://github.com/alamb/arrow-datafusion/pull/14) of how to support `InList` expressions directly by using the interval analysis library.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "Weijun-H (via GitHub)" <gi...@apache.org>.
Weijun-H commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1365870176
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
Review Comment:
```suggestion
/// Return true if the expr can be pruned by the bloom filter
```
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
+ match expr.op() {
+ Operator::And => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left || right
+ }
+ Operator::Or => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left && right
+ }
Review Comment:
```suggestion
Operator::And | Operator::Or => {
let left = Self::prune_expr_with_bloom_filter(
expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
column_sbbf,
);
let right = Self::prune_expr_with_bloom_filter(
expr.right()
.as_any()
.downcast_ref::<phys_expr::BinaryExpr>(),
column_sbbf,
);
match expr.op() {
Operator::And => left || right,
Operator::Or => left && right,
_ => false,
}
}
```
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -246,6 +250,18 @@ impl ParquetExec {
.unwrap_or(config_options.execution.parquet.enable_page_index)
}
+ /// If enabled, the reader will read the bloom filter
Review Comment:
```suggestion
/// If enabled, the reader will read by the bloom filter
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support Read bloom Filter with parquet [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1365751727
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -846,4 +1120,185 @@ mod tests {
let execution_props = ExecutionProps::new();
create_physical_expr(expr, &df_schema, schema, &execution_props).unwrap()
}
+
+ #[tokio::test]
+ async fn test_row_group_bloom_filter_pruning_predicate_simple_expr() {
+ // load parquet file
+ let testdata = datafusion_common::test_util::parquet_test_data();
+ let file_name = "data_index_bloom_encoding_stats.parquet";
+ let path = format!("{testdata}/{file_name}");
+ let data = bytes::Bytes::from(std::fs::read(path).unwrap());
+
+ // generate pruning predicate
+ let schema = Schema::new(vec![Field::new("String", DataType::Utf8, false)]);
+ let expr = col(r#""String""#).eq(lit("Hello_Not_Exists"));
+ let expr = logical2physical(&expr, &schema);
+ let pruning_predicate =
+ PruningPredicate::try_new(expr, Arc::new(schema)).unwrap();
+
+ let row_groups = vec![0];
+ let pruned_row_groups = test_row_group_bloom_filter_pruning_predicate(
+ file_name,
+ data,
+ &pruning_predicate,
+ &row_groups,
+ )
+ .await
+ .unwrap();
+ assert!(pruned_row_groups.is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_row_group_bloom_filter_pruning_predicate_partial_expr() {
+ // load parquet file
+ let testdata = datafusion_common::test_util::parquet_test_data();
+ let file_name = "data_index_bloom_encoding_stats.parquet";
+ let path = format!("{testdata}/{file_name}");
+ let data = bytes::Bytes::from(std::fs::read(path).unwrap());
+
+ // generate pruning predicate
+ let schema = Schema::new(vec![Field::new("String", DataType::Utf8, false)]);
+ let expr = col(r#""String""#)
Review Comment:
I wonder how much additional coverage that this test and the next one (`test_row_group_bloom_filter_pruning_predicate_partial_expr` and `test_row_group_bloom_filter_pruning_predicate_mutiple_expr`) are adding?
In other words, I think only one is required
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
Review Comment:
I really like how this interface is similar to [`PruningPredicate` ](https://docs.rs/datafusion/latest/datafusion/physical_optimizer/pruning/struct.PruningPredicate.html) and encapsulates the logic well.
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -846,4 +1120,185 @@ mod tests {
let execution_props = ExecutionProps::new();
create_physical_expr(expr, &df_schema, schema, &execution_props).unwrap()
}
+
Review Comment:
Here is the data in case anyone was wondering:
```sql
β― select * from 'data_index_bloom_encoding_stats.parquet';
+-----------+
| String |
+-----------+
| Hello |
| This is |
| a |
| test |
| How |
| are you |
| doing |
| today |
| the quick |
| brown fox |
| jumps |
| over |
| the lazy |
| dog |
+-----------+
14 rows in set. Query took 0.003 seconds.
```
##########
datafusion/core/src/datasource/physical_plan/parquet.rs:
##########
@@ -84,6 +84,9 @@ pub struct ParquetExec {
/// Override for `Self::with_enable_page_index`. If None, uses
/// values from base_config
enable_page_index: Option<bool>,
+ /// Override for `Self::with_enable_bloom_filter`. If None, uses
+ /// values from base_config
+ enable_bloom_filter: Option<bool>,
Review Comment:
I think we should also add a session level configuration for `enable_bloom_filter` (defaults to on), as we do for `enable_page_index` so that users have the ability to turn it off.
https://github.com/apache/arrow-datafusion/blob/c97a0d2ebb8697855edd5b56904accbfa9a80290/datafusion/common/src/config.rs#L284
We can do this as a follow on PR (it doesn't have to be this one)
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
+ match expr.op() {
+ Operator::And => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left || right
+ }
+ Operator::Or => {
+ let left = Self::prune_expr_with_bloom_filter(
+ expr.left().as_any().downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ let right = Self::prune_expr_with_bloom_filter(
+ expr.right()
+ .as_any()
+ .downcast_ref::<phys_expr::BinaryExpr>(),
+ column_sbbf,
+ );
+ left && right
+ }
+ Operator::Eq => {
+ if let Some((col, val)) = Self::check_expr_is_col_equal_const(expr) {
+ if let Some(sbbf) = column_sbbf.get(col.name()) {
+ match val {
+ ScalarValue::Utf8(Some(v)) => !sbbf.check(&v.as_str()),
+ ScalarValue::Boolean(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Float32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int64(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int32(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int16(Some(v)) => !sbbf.check(&v),
+ ScalarValue::Int8(Some(v)) => !sbbf.check(&v),
+ _ => false,
+ }
+ } else {
+ false
+ }
+ } else {
+ false
+ }
+ }
+ _ => false,
+ }
+ }
+
+ fn get_predicate_columns(
+ expr: Option<&phys_expr::BinaryExpr>,
+ ) -> Option<HashSet<String>> {
+ match expr {
Review Comment:
I think you could write this code to be more efficient by using a [`TreeNode`](https://docs.rs/datafusion/latest/datafusion/common/tree_node/trait.TreeNode.html) recursion instead. Here is a proposal: https://github.com/openobserve/arrow-datafusion/pull/5
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
+ fn prune_expr_with_bloom_filter(
+ expr: Option<&phys_expr::BinaryExpr>,
+ column_sbbf: &HashMap<String, Sbbf>,
+ ) -> bool {
+ if expr.is_none() {
+ return false;
+ }
+ let expr = expr.unwrap();
Review Comment:
this will panic if an expression like `x LIKE 'foo'` is passed, maybe it should return `false` instead like the following (or perhaps have a comment about why this is guaranteed to be `Some`)
```suggestion
let Some(expr) = expr else {
// unsupported predicate
return false;
}
```
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -846,4 +1120,185 @@ mod tests {
let execution_props = ExecutionProps::new();
create_physical_expr(expr, &df_schema, schema, &execution_props).unwrap()
}
+
Review Comment:
Here is the data in case anyone was wondering:
```sql
β― select * from 'data_index_bloom_encoding_stats.parquet';
+-----------+
| String |
+-----------+
| Hello |
| This is |
| a |
| test |
| How |
| are you |
| doing |
| today |
| the quick |
| brown fox |
| jumps |
| over |
| the lazy |
| dog |
+-----------+
14 rows in set. Query took 0.003 seconds.
```
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
+ )),
+ }
+ }
+
+ fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
+ Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ }
+
+ /// filter the expr with bloom filter return true if the expr can be pruned
Review Comment:
```suggestion
/// filter the expr with bloom filter return true if the `expr` can be proved not `true`
/// based on the bloom filter.
```
##########
datafusion/sqllogictest/test_files/json.slt:
##########
@@ -58,8 +58,10 @@ AggregateExec: mode=Final, gby=[], aggr=[COUNT(*)]
------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
--------JsonExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/2.json]]}, projection=[a]
-query error DataFusion error: Schema error: No field named mycol\.
+query N
Review Comment:
FWIW this change is related to the arrow update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "Ted-Jiang (via GitHub)" <gi...@apache.org>.
Ted-Jiang commented on code in PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#discussion_r1366386102
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Sorry i already handle in here https://github.com/apache/arrow-datafusion/blob/3f8c51222fa0aa5bb4ef7fa5cab55875bfde0a63/datafusion/core/src/physical_optimizer/pruning.rs#L759-L76 π€£ maybe we can add a ref here
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Sorry i already handle in here https://github.com/apache/arrow-datafusion/blob/3f8c51222fa0aa5bb4ef7fa5cab55875bfde0a63/datafusion/core/src/physical_optimizer/pruning.rs#L759-L76 π€£ maybe we can add a ref here
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -92,6 +98,242 @@ pub(crate) fn prune_row_groups(
filtered
}
+/// Prune row groups by bloom filters
+///
+/// Returns a vector of indexes into `groups` which should be scanned.
+///
+/// If an index is NOT present in the returned Vec it means the
+/// predicate filtered all the row group.
+///
+/// If an index IS present in the returned Vec it means the predicate
+/// did not filter out that row group.
+pub(crate) async fn prune_row_groups_by_bloom_filters<
+ T: AsyncFileReader + Send + 'static,
+>(
+ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ row_groups: &[usize],
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+) -> Vec<usize> {
+ let bf_predicates = match BloomFilterPruningPredicate::try_new(predicate.orig_expr())
+ {
+ Ok(predicates) => predicates,
+ Err(_) => {
+ return row_groups.to_vec();
+ }
+ };
+ let mut filtered = Vec::with_capacity(groups.len());
+ for idx in row_groups {
+ let rg_metadata = &groups[*idx];
+ // get all columns bloom filter
+ let mut column_sbbf =
+ HashMap::with_capacity(bf_predicates.required_columns.len());
+ for column_name in bf_predicates.required_columns.iter() {
+ let column_idx = match rg_metadata
+ .columns()
+ .iter()
+ .enumerate()
+ .find(|(_, column)| column.column_path().string().eq(column_name))
+ {
+ Some((column_idx, _)) => column_idx,
+ None => continue,
+ };
+ let bf = match builder
+ .get_row_group_column_bloom_filter(*idx, column_idx)
+ .await
+ {
+ Ok(bf) => match bf {
+ Some(bf) => bf,
+ None => {
+ continue;
+ }
+ },
+ Err(e) => {
+ log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ continue;
+ }
+ };
+ column_sbbf.insert(column_name.to_owned(), bf);
+ }
+ if bf_predicates.prune(&column_sbbf) {
+ metrics.row_groups_pruned.add(1);
+ continue;
+ }
+ filtered.push(*idx);
+ }
+ filtered
+}
+
+struct BloomFilterPruningPredicate {
+ /// Actual pruning predicate (rewritten in terms of column min/max statistics)
+ predicate_expr: Option<phys_expr::BinaryExpr>,
+ /// The statistics required to evaluate this predicate
+ required_columns: Vec<String>,
+}
+
+impl BloomFilterPruningPredicate {
+ fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
+ let expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
+ match Self::get_predicate_columns(expr) {
+ Some(columns) => Ok(Self {
+ predicate_expr: expr.cloned(),
+ required_columns: columns.into_iter().collect(),
+ }),
+ None => Err(DataFusionError::Execution(
+ "BloomFilterPruningPredicate only support binary expr".to_string(),
Review Comment:
Sorry, already handle in here https://github.com/apache/arrow-datafusion/blob/3f8c51222fa0aa5bb4ef7fa5cab55875bfde0a63/datafusion/core/src/physical_optimizer/pruning.rs#L759-L76 π€£ maybe we can add a ref here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1776024081
FWI DataFusion has been upgraded to arrow 48 now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "hengfeiyang (via GitHub)" <gi...@apache.org>.
hengfeiyang commented on PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821#issuecomment-1776780352
@alippai The `count` or `distinct` can't optimize by `bloom filter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Use bloom filter when reading parquet to skip row groups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #7821:
URL: https://github.com/apache/arrow-datafusion/pull/7821
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org