Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

parquet: Add support for row group pruning on FixedSizeBinary #9646

Merged
merged 2 commits into from
Mar 19, 2024
Merged

parquet: Add support for row group pruning on FixedSizeBinary #9646

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
496ba9a
Add support for row group pruning on FixedSizeBinary
progval Mar 16, 2024
8e11cdf
Check statistics values are valid for their type
progval Mar 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,7 @@ impl PruningStatistics for BloomFilterStatistics {
match value {
ScalarValue::Utf8(Some(v)) => sbbf.check(&v.as_str()),
ScalarValue::Binary(Some(v)) => sbbf.check(v),
ScalarValue::FixedSizeBinary(_size, Some(v)) => sbbf.check(v),
ScalarValue::Boolean(Some(v)) => sbbf.check(v),
ScalarValue::Float64(Some(v)) => sbbf.check(v),
ScalarValue::Float32(Some(v)) => sbbf.check(v),
Expand Down
27 changes: 25 additions & 2 deletions datafusion/core/src/datasource/physical_plan/parquet/statistics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,21 +105,44 @@ macro_rules! get_statistic {
let s = std::str::from_utf8(s.$bytes_func())
.map(|s| s.to_string())
.ok();
if s.is_none() {
log::debug!(
"Utf8 statistics is a non-UTF8 value, ignoring it."
);
}
Some(ScalarValue::Utf8(s))
}
}
}
// type not supported yet
// type not fully supported yet
ParquetStatistics::FixedLenByteArray(s) => {
match $target_arrow_type {
// just support the decimal data type
// just support specific logical data types, there are others each
// with their own ordering
Some(DataType::Decimal128(precision, scale)) => {
Some(ScalarValue::Decimal128(
Some(from_bytes_to_i128(s.$bytes_func())),
*precision,
*scale,
))
}
Some(DataType::FixedSizeBinary(size)) => {
let value = s.$bytes_func().to_vec();
let value = if value.len().try_into() == Ok(*size) {
Some(value)
} else {
log::debug!(
"FixedSizeBinary({}) statistics is a binary of size {}, ignoring it.",
size,
value.len(),
);
None
};
Some(ScalarValue::FixedSizeBinary(
Copy link
Contributor

@alamb alamb Mar 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we verify that the resulting bytes have the correct size? As in verify that *size is the same as (s.$bytes_func().to_vec().len()?

If not, I suggest we ignore the error (but don't use the statistics) the same way as is done for non UTF8 values i string statistics. Perhaps we can at least debug when this happens with debug!

Since this data can come from any arbitrary parquet writer, I think it is good practice to validate what is in there

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh sure, good catch.

The code for strings is:

let s = std::str::from_utf8(s.$bytes_func())
    .map(|s| s.to_string())
    .ok();
Some(ScalarValue::Utf8(s))

which looks like it's treated as a NULL, rather than ignored as a statistics value. Should I do the same here?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I think converting to NULL is the correct behavior. In this case that results in ignoring the value in terms of range analysis. More details are here if you are curious:

https://github.com/apache/arrow-datafusion/blob/8074ca1e758470319699a562074290906003b312/datafusion/core/src/physical_optimizer/pruning.rs#L85-L87

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, done. I also added a log statement for strings, for consistency.

*size,
value,
))
}
_ => None,
}
}
Expand Down
101 changes: 101 additions & 0 deletions datafusion/core/tests/parquet/row_group_pruning.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -948,6 +948,107 @@ async fn prune_binary_lt() {
.await;
}

#[tokio::test]
async fn prune_fixedsizebinary_eq_match() {
RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize = ARROW_CAST(CAST('fe6' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
// false positive on 'all frontends' batch: 'fe1' < 'fe6' < 'fe7'
.with_matched_by_stats(Some(2))
.with_pruned_by_stats(Some(1))
.with_matched_by_bloom_filter(Some(1))
.with_pruned_by_bloom_filter(Some(1))
.with_expected_rows(1)
.test_row_group_prune()
.await;

RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize = ARROW_CAST(CAST('fe6' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
// false positive on 'all frontends' batch: 'fe1' < 'fe6' < 'fe7'
.with_matched_by_stats(Some(2))
.with_pruned_by_stats(Some(1))
.with_matched_by_bloom_filter(Some(1))
.with_pruned_by_bloom_filter(Some(1))
.with_expected_rows(1)
.test_row_group_prune()
.await;
}

#[tokio::test]
async fn prune_fixedsizebinary_eq_no_match() {
RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize = ARROW_CAST(CAST('be9' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
// false positive on 'mixed' batch: 'be1' < 'be9' < 'fe4'
.with_matched_by_stats(Some(1))
.with_pruned_by_stats(Some(2))
.with_matched_by_bloom_filter(Some(0))
.with_pruned_by_bloom_filter(Some(1))
.with_expected_rows(0)
.test_row_group_prune()
.await;
}

#[tokio::test]
async fn prune_fixedsizebinary_neq() {
RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize != ARROW_CAST(CAST('be1' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
.with_matched_by_stats(Some(3))
.with_pruned_by_stats(Some(0))
.with_matched_by_bloom_filter(Some(3))
.with_pruned_by_bloom_filter(Some(0))
.with_expected_rows(14)
.test_row_group_prune()
.await;
}

#[tokio::test]
async fn prune_fixedsizebinary_lt() {
RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize < ARROW_CAST(CAST('be3' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
// matches 'all backends' only
.with_matched_by_stats(Some(1))
.with_pruned_by_stats(Some(2))
.with_matched_by_bloom_filter(Some(0))
.with_pruned_by_bloom_filter(Some(0))
.with_expected_rows(2)
.test_row_group_prune()
.await;

RowGroupPruningTest::new()
.with_scenario(Scenario::ByteArray)
.with_query(
"SELECT name, service_fixedsize FROM t WHERE service_fixedsize < ARROW_CAST(CAST('be9' AS bytea), 'FixedSizeBinary(3)')",
)
.with_expected_errors(Some(0))
.with_matched_by_stats(Some(2))
.with_pruned_by_stats(Some(1))
.with_matched_by_bloom_filter(Some(0))
.with_pruned_by_bloom_filter(Some(0))
// all backends from 'mixed' and 'all backends'
.with_expected_rows(8)
.test_row_group_prune()
.await;
}

#[tokio::test]
async fn prune_periods_in_column_names() {
// There are three row groups for "service.name", each with 5 rows = 15 rows total
Expand Down
Loading