Table Scan: Add Row Group Skipping

[sdd]

This PR, and it's follow-up, #565, introduces two more advanced features that can improve performance when executing table reads on parquet-backed tables when using a filter predicate: "row group filtering" and "row selection skipping".

Background

When performing a predicate filtered table scan, the best performance is achieved by eliminating as many rows as possible without having to apply the filter predicate to each individual row. Applying a filter predicate to the data itself takes time, and there are techniques that we can employ in order to avoid this.
These techniques are applied hierarchically, with the most coarse-grained applying first. Each step eliminates some work for the next step by excluding more rows from the result.

At the top level, when we create the file plan for the scan, we use the ManifestEvaluator to reject entire manifest files from the plan, based on the partition field summary metadata of each manifest file. Rejecting a manifest file allows us to filter out multiple, often many, data files, in a single step.

Once we have our list of remaining selected manifest files, we load them in and perform the next layer of filtering. In this step, we use the ExpressionEvaluator to apply our filter predicate to the partition summary of each entry in the manifest files. Each entry in the manifest file corresponds to a single data file, and ExpressionEvaluator allows us to reject individual data files from the scan based on their partition summary metadata within each manifest entry.

The final filter step in the file plan stage uses the InclusiveMetricsEvaluator. This evaluates our filter predicate against the field-level metrics that are stored in the manifest entry. These contain a variety of statistics for each field in the data file: the number of nulls, NaNs, and values, and the lower and upper bounds for the field (i.e. lowest / highest value for numerical columns, or the alphabetically first and last values for string columns). This allows us to reject even more of the remaining data files before we've even scanned them.

All of this is already in place prior to this PR. The general theme is to use metadata in order to reject or select chunks of data before we even get to the data itself. We don't need to stop this process at the data file level though - Parquet files are not monolithic, but chunked themselves, and contain two sources of metadata that we can use to further reject chunks of data from our scan before we even look at it.

Parquet files split the data within into Row Groups. The data within each row group is columnar - i.e. each column is stored separately, one after the other, in "Column Chunks". The data for each column chunk in a row group can then be further subdivided into pages. The size of each page does not need to be the same, either within a column or across columns: i.e., column 1, page 1 does not necessarily have the same number of rows as either column 1, page 2 or column 2, page 1.

The footer of the Parquet file stores metadata about each row group. For each row group, there are some statistics that are maintained for every column chunk in that row group. These include the min and max value present in that column chunk and the number of nulls - similar to the data file metrics that get processed by the InclusiveMetricsEvaluator mentioned above.

Optionally (but always for Iceberg Parquet files), the column chunk metadata can also contain a Page Index. This extra metadata contains further statistics about each page within the column chunk: specifically, the min and max value within each page of the chunk, and how many null values are in that page. This final level of metadata can be used to further filter out rows from each row group that can't match the filter predicate.

Row Group Filtering

This PR adds the RowGroupMetricsEvaluator, which applies an Iceberg filter predicate to a parquet file's row group metadata as outlined above to reject or select whole row groups within the data file from the scan. The ArrowReader uses this within calls to read, passing the list of selected row group indices to ParquetRecordBatchStreamBuilder::with_row_groups.

Since the row group metadata is always parsed anyway as part of the process of reading each parquet file, this filtering adds little overhead. However, it is only useful if your table's data files can have more than one row group in each data file. The default max row group size for Iceberg parquet writing is 128Mb (configured using the write.parquet.row-group-size-bytes table setting, see here). If your data files are smaller than the row group size, then Row Group Filtering will provide no performance improvements. This PR makes Row Group Filtering enabled by default, but it can be turned off for a scan by calling with_row_group_filtering_enabled(false) when building the scan (There's a corresponding method on the ArrowReaderBuilder if you are using that directly). You may want to do this to eliminate the small but non-zero overhead of this filtering, or when experimenting to see the impact that this filtering can make.

Future Directions

Row Selection Skipping has been moved from this PR into #565.

The overhead of parsing the Page Index could be offset by caching a parsed Page Index for each data file. It would be interesting to see if this has enough of an effect to increase the utility of Row Selection Skipping. I will experiment with adding Page Index caching to my Object Cache PR to see how this goes.

Additionally, the page index is an array, with one index per column. It seems like it would be possible to only store page index entries for some arbitrary left-aligned slice of columns - i.e., columns 0 to m for a file with n columns, where m < n. This could be taken advantage of, in conjunction with using a specific column order, to only store a page index for columns for which it would be useful to filter on, saving space and potentially reducing the parsing overhead.

[liurenjie1024]

Thanks @sdd for this pr. I just skimmed through it and thought that it's great! My suggestion is to split this pr into two smaller ones: one for row group and one for page index. You can see that without tests this pr is already nearly 1.6k+ lines, and it would be too large to review.

[sdd]

Hey @liurenjie1024 - I've removed the row selection change and will raise that in a follow-up PR instead.

[liurenjie1024]

Thanks @sdd for this great pr! I've left some suggestions, but overall it's great!

[liurenjie1024]

I don't think it's safe to use min_bytes/max_bytes method here, as it's parquet's internal binary format. We should use min/max method value with primitive type to convert it to Datum. There are two point to notice here:

1. min/max value may not be set here, we need to check has_min_max_set to avoid panic
2. min/max may not be exact, they maybe truncated for binary values, we should check min_is_exact/max_is_exact method to ensure that.

[sdd]

I've updated (not via a force-push 😅 ) the PR to make use of min and max rather than min_bytes and max_bytes, as well as using has_min_max_set. I've covered all but the most obscure primitive types, as we've done similarly in other places.

min/max may not be exact, they maybe truncated for binary values, we should check min_is_exact/max_is_exact method to ensure that.

What should the behaviour be when min_is_exact() == false / max_is_exact() == false?

[liurenjie1024]

I think it should be similar to stat in iceberg. Stat in iceberg is also truncated.

[liurenjie1024]

It would be great if we have some tests to verify it.

[sdd]

Yep - in progress, should not take too long.

[sdd]

Back to you, @liurenjie1024 - I've added tests to cover (hopefully!) all code paths in the RowGroupMetricsEvaluator.

[liurenjie1024]

Thanks @sdd for this great pr!