Table Scan: Add Row Selection Filtering

[sdd]

Row Selection Skipping

This PR adds the PageIndexEvaluator, which applies an iceberg filter predicate to a parquet Page Index, allowing the fetching into memory and decoding of entire slices of rows within a row group to be skipped if the PageIndexEvaluator determines from the Page Index that they don't match the filter. The PageIndexEvaluator builds a RowSelection that it passes to ParquetRecordBatchStreamBuilder::with_row_selection.

Unlike the row group metadata, the Page Index is not parsed by default when reading in the parquet file metadata. When present, especially for files with many columns, it can be quite large and intensive to parse. In fact, during experiments on the performance testing suite on my other PR, it turns out that often the additional time required to parse the Page Index outweighs the time saved fetching, decoding, and filtering all of the rows within a row group. For this reason, I've kept Row Selection Skipping disabled by default. It can be enabled by calling with_row_selection_enabled(true) when building the scan (There's a corresponding method on the ArrowReaderBuilder if you are using that directly). If you're interested in this feature, you may want to experiment with the write.parquet.page-size-bytes and write.parquet.page-row-limit table settings (which default to 1Mb and 20,000 rows respectively) in order to get optimal performance, as well as the sort settings for your table.

More info can be found in the preceding PR to this one, #558

[sdd]

@liurenjie1024 / @Xuanwo - PTAL when you get chance, thanks 👍🏼

[liurenjie1024]

Thanks @sdd for this great pr! Genarally LGTM, and I've left some minor points to improve, but overall it looks good!

[liurenjie1024]

I think this should be computed only once?

[sdd]

Good spot. I think the optimal approach here would not be to calc row counts for all columns up-front because only the ones for columns that appear in the predicate would be needed. Caching the result would make more sense. If it is ok with you I'll add a TODO for this rather than address in this PR, and we can address in a follow-up, as it is not broken and caching would not save us a huge amount of time here.

[liurenjie1024]

Sounds reasonable to me.

[liurenjie1024]

This is correct, but java's implementation follows jdk's default behavior: https://docs.oracle.com/javase/8/docs/api/java/lang/Double.html#compareTo-java.lang.Double-

I think in future we could unify the logic of datum comparison, but it doesn't have to be a blocker here.

[sdd]

Let's defer then considering we want this in a 0.4.0 release.

[sdd]

Back to you @liurenjie1024, thanks for the thorough review - some good spots! 😄

[liurenjie1024]

Thanks @sdd for this great pr!

[liurenjie1024]

Sounds reasonable to me.