[FEA] Enable Page-level filtering based on the ColumnIndex feature from parquet 1.11

[revans2]

Is your feature request related to a problem? Please describe.
The high level goal is to be able to reduce the amount of data that we read from disk on parquet files with a ColumnIndex/PageIndex.

https://github.com/apache/parquet-format/blob/master/PageIndex.md

Describe the solution you'd like

In Spark we currently do some hacked up things when reading parquet files. We use parquet-mr to read the metadata about the file(s). We then let it do a predicate push down to find the row groups that fit the requested predicate. Finally we read the pages for the desired columns in those row groups and put it all back together as a new in-memory parquet file that we send down to CUDF. It is ugly, but it let us do add a lot of features in Spark before CUDF could support them. It still lets us read the data using the existing Hadoop File System interface, which because it is a "standard" that our customers can and do replace. We might be able to move to the Arrow FileSystem API, but I'll talk about that in the alternatives.

Ideally I would like an API where we can send compressed pages and metadata to CUDF for decoding. The metadata would include things like the file and row group that the pages came from and what range of column indicies within those pages we would like to be read.

The hard part with filtering using a ColumnIndex is that the pages within a row group are not split on the same row boundaries, like a row group is. An example might help here. Lets say we have two columns A and B in a row group. The predicate to push down is A > 100, which corresponds to page 5 in the row group. That page is for rows 500-599. Column B requires us to load 2 pages to cover that same range. In this case lets say pages 10 and 11 which cover 450 - 549 and 550 to 700 respectively. So we would have to hand CUDF the pages 5, 10, and 11 along with the metadata about the row group and file so CUDF can know how to decode the data, and information to say only decode the rows 500 to 599 and throw away anything else that is outside of that. In a real situation it is probably going to be a lot more complicated.

Ideally this would let us pass down row groups from multiple different files too. I am not 100% sure how the row group filtering works on a multi-file source_info.

Describe alternatives you've considered
The other alternative is for us to start using the Arrow FileSystem API and also have cudf implement row number filtering (need to check, but I think the row numbers are relative to the start of the row group and not total within the file) similar to the set_row_groups API that currently exists.

This is kind of hard for us to do.

1. One of the main performance features that we have is overlapping I/O with computation. Spark likes to use lots of threads, more than we want to allow on the GPU at any point in time for memory reasons. So we let some onto the GPU, but we let the others read data to CPU memory. This lets us overlap the slow reading from a remote file system with computation on the GPU. We would need some kind of callback, or multiple APIs so that we could have CUDF read all of the needed data into host memory, and then we can wait until it is time to run at which point we can finish processing the data.
2. JNI is very slow for moving data. We use a number of tricks/alternative APIs to get around this. This is especially true when calling back from native code into Java. So having an API that goes from java to C back to java so it can read data over a socket (through C again, but a slightly more optimized interface than JNI) is far from ideal especially if it is going to involve small reads.

Additional context
This is related to #9268, but the read side compared to the write side of it.

[devavret]

I'm trying to understand this. So there's broadly two steps where parquet reader spends time:

1. Reading rowgroup data from the file.
2. Decompressing and decoding said data.

In case of rowgroup based filtering, you can ask cudf for the rowgroups to read which means rest of the rowgroups are neither loaded from disk into host/device memory nor decompressed/decoded.
You can filter based on exact row numbers but that would mean the entire rowgroup is still read from disk but only the necessary pages are decompressed/decoded. So only partial time savings.

With this new page level filtering, you're trying to avoid reading unnecessary pages from within the rowgroups, from disk. Is that correct?

[github-actions]

This issue has been labeled inactive-30d due to no recent activity in the past 30 days. Please close this issue if no further response or action is needed. Otherwise, please respond with a comment indicating any updates or changes to the original issue and/or confirm this issue still needs to be addressed. This issue will be labeled inactive-90d if there is no activity in the next 60 days.

[github-actions]

This issue has been labeled inactive-90d due to no recent activity in the past 90 days. Please close this issue if no further response or action is needed. Otherwise, please respond with a comment indicating any updates or changes to the original issue and/or confirm this issue still needs to be addressed.

[GregoryKimball]

API where we can send compressed pages and metadata to CUDF for decoding. The metadata would include things like the file and row group that the pages came from and what range of column indicies within those pages we would like to be read.

@revans2 would you please share more about the API you have in mind? Would the inputs be a new metadata class, or are there existing ones that would suffice? Would the output be a table or an iterable of column fragments?

[revans2]

Sorry about the long delay for a response. I see two possible APIs.

The first would be that in Spark we add in a glue layer that would let the CUDF reader call back into java and the Hadoop File System API so we can properly support security and all of the various pluggable APIs that Spark/Java can support, but CUDF does not yet.

CUDF would then need to add in APIs that would allow us to partition the data the same way that Spark does, not hard I can show you how to do it. APIs to do predicate push down so row groups can be skipped if needed (and in this case possibly pages within a row group being skipped too). APIs to do column pruning so we can tell you exactly which columns we want read. A callback API so we know just before data is moved to the GPU so we can overlap I/O as much as possible, and not overload the GPU's memory. And an API, if it does not exist already, to be able to group multiple smaller files together so that the GPU can be more efficient when reading the data. Also we can use a thread pool to read data from some file systems, because the bandwidth for a lot of CSP blob stores is limited by each connection or single threaded CPU performance in decompressing/decrypting the data, so having more connections improves the total throughput.

I like this approach because it allows for more improvement in the future, including things like GPU direct storage and it bypasses having to read/write the footer multiple times. But we would have to do a bunch of profiling to be sure that the java callback is efficient and not slowing things down, and it would be a lot of large changes to CUDF to add in all of the features that we have been doing directly in scala today.

The second approach would be something where we would pass down structured data to CUDF. So instead of passing down a buffer that holds a complete parquet file. We would parse the footer and the ColumnIndex ourselves. Read the data that we know we need to read, and then pass the information to CUDF as almost a tree, where we would give you the some form of a footer that would point to the columns, row groups, pages, and row ranges that we care about and want read.

This approach is a lot more complicated to do, in my option, and is prone to also being slow because sending structured data through JNI is also rather problematic and slow. So I would prefer the first option, but...