Implement cudf-polars chunked parquet reading

[brandon-b-miller]

This PR provides access to the libcudf chunked parquet reader through the cudf-polars gpu engine, inspired by the cuDF python implementation.

Closes #16818

[wence-]

Minor tweaks, overall, I think the implementation side looks in good shape.

There's an open question about whether we want to read all the chunks and do a concatenate at the end, but I will wait for benchmarking on that.

[wence-]

We need to put some documentation about the chunked reading somewhere. As a minimum, can you add a new md file in cudf_polars/docs and talk through what settings are available, and what they mean?

[brandon-b-miller]

In addition to the slicing issue, switching to chunked reading by default seems to shake out another chunked parquet reader bug with n_rows > 0

import pylibcudf as plc
import pyarrow as pa
import pyarrow.parquet as pq


data = {
            "a": [1, 2, 3, None, 4, 5],
            "b": ["ẅ", "x", "y", "z", "123", "abcd"],
            "c": [None, None, 4, 5, -1, 0],
        }


path = "./test.parquet"
pq.write_table(pa.Table.from_pydict(data), path)

reader = plc.io.parquet.ChunkedParquetReader(
        plc.io.SourceInfo([path]),
        columns=['a', 'b', 'c'],
        nrows=2,
        skip_rows=0,
        chunk_read_limit=0,
        pass_read_limit=17179869184 # 16 GiB
)
chk = reader.read_chunk()
tbl = chk.tbl
names = chk.column_names()
concatenated_columns = tbl.columns()
while reader.has_next():
    tbl = reader.read_chunk().tbl

    for i in range(tbl.num_columns()):
        concatenated_columns[i] = plc.concatenate.concatenate(
            [concatenated_columns[i], tbl._columns[i]]
        )
        # Drop residual columns to save memory
        tbl._columns[i] = None

gpu_result = plc.interop.to_arrow(tbl)
cpu_result = pq.read_table(path)[:2]

print(cpu_result.column(1).to_pylist())
print(gpu_result.column(1).to_pylist())

this yields

['ẅ', 'x']
['ẅ', 'x\x00\x00\x00\x00\x00\x00\x00\x00\x00']

I suppose this is a separate issue from #17158 since the trunk includes this fix now I believe.

[wence-]

I don't like this, because it turns off yet more routes into running a query on device. Particularly, since parquet ingest is the primary way we recommend people do things, we need it to work in basically all cases.

Things we can do:

• turn off chunked reading is skip_rows != 0
• fail if skip_rows != 0 (as here)
• Manually handle skip_rows != 0 in the chunked reader by reading full chunks and slicing them away if skip_rows != 0

I think I like the third option best.

[wence-]

OK, I pushed an implementation of option 3.

[vyasr]

I have some suggestions to improve the docs and a couple of minor questions on implementation, but nothing blocking. I think that this is good to merge when you are happy.

[brandon-b-miller]

/merge

[rjzamora]

@wence- - Just a note. Pretty sure this means we will need to pass in a config object to every single task in the task graph for multi-gpu.

[wence-]

Argh, ok, painful. Let's try and figure something out (especially because the config object can contain a memory resource).

[rjzamora]

Would it make sense for config to be a required IR constructor argument, and not require it as an argument to do_evaluate (unless necessary)?

[wence-]

Ah, we could pass the config options we need into the Scan node during translate, and then it never needs to be in do_evaluate at all

[rjzamora]

Possible revision proposed here: #17339