You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "alamb (via GitHub)" <gi...@apache.org> on 2023/12/06 18:04:20 UTC
[PR] Rewrite bloom filters to use contains API [arrow-datafusion]
alamb opened a new pull request, #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442
Draft
- [ ] Builds on https://github.com/apache/arrow-datafusion/pull/8440
## Which issue does this PR close?
Closes https://github.com/apache/arrow-datafusion/issues/8376
Closes https://github.com/apache/arrow-datafusion/pull/8397
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
We introduced contains in https://github.com/apache/arrow-datafusion/pull/8440, and now we can use it to simplify the bloom filter code
## Are these changes tested?
Covered by existing tests (including that added in https://github.com/apache/arrow-datafusion/pull/8433)
## 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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "waynexia (via GitHub)" <gi...@apache.org>.
waynexia commented on code in PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#discussion_r1435933829
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -123,182 +116,116 @@ pub(crate) async fn prune_row_groups_by_bloom_filters<
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 literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) = parquet_column(
+ builder.parquet_schema(),
+ predicate.schema(),
+ &column_name,
+ ) else {
+ continue;
};
+
let bf = match builder
.get_row_group_column_bloom_filter(*idx, column_idx)
.await
{
- Ok(bf) => match bf {
- Some(bf) => bf,
- None => {
- continue;
- }
- },
+ Ok(Some(bf)) => bf,
+ Ok(None) => continue, // no bloom filter for this column
Err(e) => {
- log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ log::debug!("Ignoring error reading bloom filter: {e}");
Review Comment:
debug level looks good to me. non-fatal errors on query path are easy to flow 👍
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#discussion_r1417769850
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -123,182 +116,116 @@ pub(crate) async fn prune_row_groups_by_bloom_filters<
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 literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) = parquet_column(
Review Comment:
This also fixes a potential bug similar to https://github.com/apache/arrow-datafusion/issues/8335 by finding the correct fields
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -123,182 +116,116 @@ pub(crate) async fn prune_row_groups_by_bloom_filters<
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 literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) = parquet_column(
+ builder.parquet_schema(),
+ predicate.schema(),
+ &column_name,
+ ) else {
+ continue;
};
+
let bf = match builder
.get_row_group_column_bloom_filter(*idx, column_idx)
.await
{
- Ok(bf) => match bf {
- Some(bf) => bf,
- None => {
- continue;
- }
- },
+ Ok(Some(bf)) => bf,
+ Ok(None) => continue, // no bloom filter for this column
Err(e) => {
- log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ log::debug!("Ignoring error reading bloom filter: {e}");
metrics.predicate_evaluation_errors.add(1);
continue;
}
};
- column_sbbf.insert(column_name.to_owned(), bf);
+ column_sbbf.insert(column_name.to_string(), bf);
}
- if bf_predicates.prune(&column_sbbf) {
+
+ let stats = BloomFilterStatistics { column_sbbf };
+
+ // Can this group be pruned?
+ let prune_result = predicate.prune(&stats);
+ let prune_group = match prune_result {
+ Ok(values) => !values[0],
+ Err(e) => {
+ log::debug!("Error evaluating row group predicate on bloom filter: {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ false
+ }
+ };
+
+ if prune_group {
metrics.row_groups_pruned.add(1);
- continue;
+ } else {
+ filtered.push(*idx);
}
- filtered.push(*idx);
}
filtered
}
-struct BloomFilterPruningPredicate {
- /// Actual pruning predicate
- predicate_expr: Option<phys_expr::BinaryExpr>,
- /// The statistics required to evaluate this predicate
- required_columns: Vec<String>,
+struct BloomFilterStatistics {
+ /// Maps column name to the parquet bloom filter
+ column_sbbf: HashMap<String, Sbbf>,
}
-impl BloomFilterPruningPredicate {
- fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
- let binary_expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
- match binary_expr {
- Some(binary_expr) => {
- let columns = Self::get_predicate_columns(expr);
- Ok(Self {
- predicate_expr: Some(binary_expr.clone()),
- required_columns: columns.into_iter().collect(),
- })
- }
- None => Err(DataFusionError::Execution(
- "BloomFilterPruningPredicate only support binary expr".to_string(),
- )),
- }
+impl PruningStatistics for BloomFilterStatistics {
+ fn min_values(&self, _column: &Column) -> Option<ArrayRef> {
+ None
}
- fn prune(&self, column_sbbf: &HashMap<String, Sbbf>) -> bool {
- Self::prune_expr_with_bloom_filter(self.predicate_expr.as_ref(), column_sbbf)
+ fn max_values(&self, _column: &Column) -> Option<ArrayRef> {
+ None
}
- /// Return true if the `expr` can be proved not `true`
- /// based on the bloom filter.
- ///
- /// We only checked `BinaryExpr` but it also support `InList`,
- /// Because of the `optimizer` will convert `InList` to `BinaryExpr`.
- fn prune_expr_with_bloom_filter(
Review Comment:
All of this logic is now split between `LiteralGuarantee` https://github.com/apache/arrow-datafusion/pull/8437 and `PruningPredicate`: https://github.com/apache/arrow-datafusion/pull/8440
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -123,182 +116,116 @@ pub(crate) async fn prune_row_groups_by_bloom_filters<
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 literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) = parquet_column(
+ builder.parquet_schema(),
+ predicate.schema(),
+ &column_name,
+ ) else {
+ continue;
};
+
let bf = match builder
.get_row_group_column_bloom_filter(*idx, column_idx)
.await
{
- Ok(bf) => match bf {
- Some(bf) => bf,
- None => {
- continue;
- }
- },
+ Ok(Some(bf)) => bf,
+ Ok(None) => continue, // no bloom filter for this column
Err(e) => {
- log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ log::debug!("Ignoring error reading bloom filter: {e}");
Review Comment:
in general, errors in logs I think are meant to be "something bad happened that caused a user error" -- in this case the query will proceed so I think debug is a better level.
If someone needs it in production logs perhaps we can change it to `log`
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#discussion_r1435592501
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -118,188 +111,130 @@ pub(crate) fn prune_row_groups_by_statistics(
pub(crate) async fn prune_row_groups_by_bloom_filters<
T: AsyncFileReader + Send + 'static,
>(
+ arrow_schema: &Schema,
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,
+ // get all columns in the predicate that we could use a bloom filter with
+ let literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) =
+ parquet_column(builder.parquet_schema(), arrow_schema, &column_name)
Review Comment:
This also fixes a potential bug similar to https://github.com/apache/arrow-datafusion/issues/8335 by finding the correct fields
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#discussion_r1417769850
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -123,182 +116,116 @@ pub(crate) async fn prune_row_groups_by_bloom_filters<
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 literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) = parquet_column(
Review Comment:
This also fixes a potential bug similar to https://github.com/apache/arrow-datafusion/issues/8335 by finding the correct fields
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#issuecomment-1868287281
This PR is now ready for review @haohuaijin and @waynexia
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#discussion_r1435602195
##########
datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs:
##########
@@ -118,188 +111,129 @@ pub(crate) fn prune_row_groups_by_statistics(
pub(crate) async fn prune_row_groups_by_bloom_filters<
T: AsyncFileReader + Send + 'static,
>(
+ arrow_schema: &Schema,
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,
+ // get all columns in the predicate that we could use a bloom filter with
+ let literal_columns = predicate.literal_columns();
+ let mut column_sbbf = HashMap::with_capacity(literal_columns.len());
+
+ for column_name in literal_columns {
+ let Some((column_idx, _field)) =
+ parquet_column(builder.parquet_schema(), arrow_schema, &column_name)
+ else {
+ continue;
};
+
let bf = match builder
.get_row_group_column_bloom_filter(*idx, column_idx)
.await
{
- Ok(bf) => match bf {
- Some(bf) => bf,
- None => {
- continue;
- }
- },
+ Ok(Some(bf)) => bf,
+ Ok(None) => continue, // no bloom filter for this column
Err(e) => {
- log::error!("Error evaluating row group predicate values when using BloomFilterPruningPredicate {e}");
+ log::debug!("Ignoring error reading bloom filter: {e}");
metrics.predicate_evaluation_errors.add(1);
continue;
}
};
- column_sbbf.insert(column_name.to_owned(), bf);
+ column_sbbf.insert(column_name.to_string(), bf);
}
- if bf_predicates.prune(&column_sbbf) {
+
+ let stats = BloomFilterStatistics { column_sbbf };
+
+ // Can this group be pruned?
+ let prune_group = match predicate.prune(&stats) {
+ Ok(values) => !values[0],
+ Err(e) => {
+ log::debug!("Error evaluating row group predicate on bloom filter: {e}");
+ metrics.predicate_evaluation_errors.add(1);
+ false
+ }
+ };
+
+ if prune_group {
metrics.row_groups_pruned.add(1);
- continue;
+ } else {
+ filtered.push(*idx);
}
- filtered.push(*idx);
}
filtered
}
-struct BloomFilterPruningPredicate {
- /// Actual pruning predicate
- predicate_expr: Option<phys_expr::BinaryExpr>,
- /// The statistics required to evaluate this predicate
- required_columns: Vec<String>,
+/// Implements `PruningStatistics` for Parquet Split Block Bloom Filters (SBBF)
+struct BloomFilterStatistics {
+ /// Maps column name to the parquet bloom filter
+ column_sbbf: HashMap<String, Sbbf>,
}
-impl BloomFilterPruningPredicate {
- fn try_new(expr: &Arc<dyn PhysicalExpr>) -> Result<Self> {
- let binary_expr = expr.as_any().downcast_ref::<phys_expr::BinaryExpr>();
- match binary_expr {
- Some(binary_expr) => {
- let columns = Self::get_predicate_columns(expr);
- Ok(Self {
- predicate_expr: Some(binary_expr.clone()),
- required_columns: columns.into_iter().collect(),
- })
- }
- None => Err(DataFusionError::Execution(
- "BloomFilterPruningPredicate only support binary expr".to_string(),
- )),
- }
+impl PruningStatistics for BloomFilterStatistics {
Review Comment:
❤
--
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] Rewrite bloom filters to use contains API [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #8442:
URL: https://github.com/apache/arrow-datafusion/pull/8442#issuecomment-1869487043
Thank you for the review @waynexia and @haohuaijin
--
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