feat:Add function for row alignment with page mask by Ted-Jiang · Pull Request #1791 · apache/arrow-rs

Ted-Jiang · 2022-06-05T04:39:03Z

Which issue does this PR close?

Closes #1790.

Rationale for this change

For now row group filter in datafusion pass a closure to arrow-rs

fn build_row_group_predicate(
    pruning_predicate: &PruningPredicate,
    metrics: ParquetFileMetrics,
) -> Box<dyn FnMut(&RowGroupMetaData, usize) -> bool> {

https://github.com/apache/arrow-datafusion/blob/585bc3a629b92ea7a86ebfe8bf762dbef4155710/datafusion/core/src/physical_plan/file_format/parquet.rs#L559-L562

So for page filter in datafusion, define filter_predicate

 Box<dyn FnMut(&[pageIndex], &[pageLocation], usize) -> &[bool]>

datafusion will send a mask(&[bool]) to arrow-rs,
then use mask call compute_row_ranges to construct RowRanges : row ranges in a row-group (one col) if col is sorted vec size will be 1.
For multi filter combine:
if there are two filters use and connect，use RowRanges::intersection to get the final rowRange; two filters use or connect，use RowRanges::union to get the final rowRange.

After this PR: i will working on column_page_reader, enable read specific row ranges record.

What changes are included in this PR?

Are there any user-facing changes?

codecov-commenter · 2022-06-05T05:15:55Z

Codecov Report

Merging #1791 (adc48cf) into master (c1a91dc) will increase coverage by 0.02%.
The diff coverage is 91.44%.

@@            Coverage Diff             @@
##           master    #1791      +/-   ##
==========================================
+ Coverage   83.39%   83.42%   +0.02%     
==========================================
  Files         198      199       +1     
  Lines       56142    56427     +285     
==========================================
+ Hits        46821    47075     +254     
- Misses       9321     9352      +31

Impacted Files	Coverage Δ
parquet/src/file/metadata.rs	`95.12% <ø> (ø)`
parquet/src/file/serialized_reader.rs	`94.46% <ø> (-1.17%)`	⬇️
parquet/src/file/page_index/range.rs	`91.44% <91.44%> (ø)`
parquet/src/util/cursor.rs	`62.18% <0.00%> (-1.69%)`	⬇️
parquet_derive/src/parquet_field.rs	`65.98% <0.00%> (-0.23%)`	⬇️
arrow/src/ipc/reader.rs	`90.73% <0.00%> (-0.11%)`	⬇️
parquet/src/file/writer.rs	`92.84% <0.00%> (-0.02%)`	⬇️
parquet/src/arrow/arrow_writer.rs	`97.76% <0.00%> (-0.02%)`	⬇️
parquet/src/arrow/schema.rs	`96.81% <0.00%> (-0.01%)`	⬇️
... and 14 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c1a91dc...adc48cf. Read the comment docs.

parquet/src/file/page_index/range.rs

Co-authored-by: Liang-Chi Hsieh <[email protected]>

Ted-Jiang · 2022-06-05T06:13:01Z

@tustvold @alamb @viirya PTAL 😊

tustvold · 2022-06-05T06:37:52Z

Thank you, I will review this tomorrow (GMT)

parquet/src/file/page_index/range.rs

viirya · 2022-06-05T07:08:21Z

parquet/src/file/page_index/range.rs

+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(


How will this be used?

Do you need to make something like with_predicate in ReadOptionsBuilder to take a closure for filtering pages based on the ranges?

Yes, basically add a closure called page_predicates in ReadOptionsBuilder like

Box<dyn FnMut(&[pageIndex], &[pageLocation], usize) -> &[bool]>

to generate mask then call this function compute_row_ranges .

I'm thinking about doing this at the end for testing with datafusion.

tustvold

Looking good, left some high level comments. Will review more thoroughly tomorrow

tustvold · 2022-06-05T08:55:03Z

parquet/src/file/metadata.rs

    num_rows: i64,
    total_byte_size: i64,
    schema_descr: SchemaDescPtr,
+    // Todo add filter result -> row range


The more I think about this the more I wonder whether the metadata structs are the right place to put the index information. They're parsed and interpreted separately from the main metadata, and so I think it makes sense for them to be stored separately?

Yes, you are right. page index stored in file-meta level.
My thought is read less pageIndex after rowgroup filter

arrow-rs/parquet/src/file/serialized_reader.rs

Lines 211 to 224 in be38829

let mut filtered_row_groups = Vec::<RowGroupMetaData>::new();

for (i, rg_meta) in row_groups.into_iter().enumerate() {

let mut keep = true;

for predicate in &mut predicates {

if !predicate(&rg_meta, i) {

keep = false;

break;

}

}

if keep {

filtered_row_groups.push(rg_meta);

}

}

arrow-rs/parquet/src/file/serialized_reader.rs

Lines 246 to 249 in be38829

metadata: ParquetMetaData::new(

metadata.file_metadata().clone(),

filtered_row_groups,

),

So i want to read index here and insert it into RowGroupMetaData.
It was just a simple idea at first, maybe we can find a better way in the process of implementation

page index stored in file-meta level.

It isn't even file-meta level, it isn't part of the footer but stored as separate pages 😅

It was just a simple idea at first, maybe we can find a better way in the process of implementation

Provided we take care to ensure we keep things pub(crate) so we don't break APIs, this seems like a good strategy 👍

It isn't even file-meta level, it isn't part of the footer but stored as separate pages 😅

yes separately from RowGroup, before the footer !😂

parquet/src/file/page_index/range.rs

parquet/src/file/page_index/mod.rs

parquet/src/file/page_index/range.rs

tustvold

I think this looks good, I am, however, a bit confused as to how it will be used... I had expected something like the following.

For each not-pruned row group:

Use Index to identify covering set of rows based on predicates
Pass row selection down to RecordReader
Add a skip_next_page to PageReader
Add a skip_values to ColumnValueDecoder
Have RecordReader use a combination of the above to skip pages and rows during decode based on the row selection

This would also naturally extend to #1191

I'm not entirely sure where the datastructure added in this PR would fit into this?

Edit: is the intention to use this for the first step?

parquet/src/file/page_index/range.rs

Co-authored-by: Raphael Taylor-Davies <[email protected]>

Ted-Jiang · 2022-06-06T12:57:24Z

Edit: is the intention to use this for the first step?

Basically, yes ! 👍

Use Index to identify covering set of rows based on predicates
Pass row selection down to RecordReader
Add a skip_next_page to PageReader
Add a skip_values to ColumnValueDecoder
Have RecordReader use a combination of the above to skip pages and rows during decode based on the row selection

First step contains:

Use SerializedFileReader read fileMeta get file_metadata(optional on page_index)
Use min_max on row group filter filter useless group (already exist)
Read specific row group page_index construct Index(Support reading PageIndex from column metadata #1761)
Use both index and filter generate pagemask (this one)
....

@tustvold I will follow up in PageReader
Hope will get big performance improvement in selective query!

Co-authored-by: Raphael Taylor-Davies <[email protected]>

Ted-Jiang added 5 commits June 3, 2022 21:00

add range and rowRanges

116fcbb

add some tests for range

0f23f20

add filter logic

e9c98c0

fix fmt

761c929

fix todo

be38829

github-actions bot added the parquet Changes to the parquet crate label Jun 5, 2022

fix test

a4c9e60