Materialize predicate columns before projection columns

[bchalk101]

Description

Background

I am currently working with a Parquet dataset, where some columns are very large ie large lists or JPEG images in string representation. The dataset size is about 80_000 rows, where each row is approx. 3mb. Another thing to note about the dataset is that it is fully randomized/shuffled, and can not be sorted, with the implication being that the statistics in each row group do not help and almost all row groups need to be read to apply simple filters.

I am trying to filter over this dataset, selecting only a few rows from the entire dataset.

An example filter is:

import polars as pl

df = pl.scan_parquet("s3://some-dataset/*.parquet").filter(pl.col("error_1") > 0.9999).collect(streaming=True)

Issue

This filter takes quite a bit of time to execute, given that it hasn't run out of memory before that.

From taking a look through the function

polars/crates/polars-io/src/parquet/read_impl.rs

Line 255 in a8bdc76

fn rg_to_dfs_par_over_rg(

it appears that all the columns are first materialized and once materialized the predicate is applied.

Suggestion

I'm proposing that that function, first loads only the predicate columns and applies the predicate, and only if the resulting data frame is not empty load all the projection columns.

If this is accepted, I have implemented this for myself and would be happy to open a PR.
(It's very rough currently, brute forced to prove out the idea.)

[coastalwhite]

This is done with the parallel=prefiltered setting.

[bchalk101]

@coastalwhite as mentioned here, #13746 (comment), prefiltered doesn't exactly address this issue.

Prefiltered will still "download" the data even if it isn't needed.

[coastalwhite]

It shouldn't. If so, I would consider that missing part of prefiltered.