Streaming Q4 implementation

[TomAugspurger]

This implements TPCH query 4 using rapidsmpf.

The primary notable thing about q4 is the left semi join / where exists between the two tables:

q = (
    # SQL exists translates to semi join in Polars API
    orders.join(
        (lineitem.filter(pl.col("l_commitdate") < pl.col("l_receiptdate"))),
        left_on="o_orderkey",
        right_on="l_orderkey",
        how="semi",
    )
    ...
)

I want to think a bit more about how to do this. Right now, I've implemented a version that broadcasts the smaller orders table and shuffles the larger lineitem table. We're currently unable to reuse the filtered joiner across chunks. cudf only supports probing with the left table (the smaller one in our case, orders). The build table must be the right table (the larger one in our case, lineitem). So I think we have to shuffle lineitem otherwise we'll get incorrect results (a key in orders matching against a key in lineitem multiple times, once per partition).

[copy-pr-bot]

Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually.

Contributors can view more details about this message here.

[TomAugspurger]

a338046 uses cudf::ast::column_name_reference instead of column_reference to avoid including o_orderdate in the output (which isn't used outside of the filter, which is done in the read_parquet now).

[wence-]

A few more comments, I think this looks pretty good now!

[wence-]

Why do you take left_chunk by const ref, but right_chunk by rvalue reference (i.e. caller must move it).

[TomAugspurger]

I'm not really sure (I'm reading up on the semantics of the two now).

We use this streaming:TableChunk&& type for the other functions (inner_join_chunk) so I suspect I was trying to match that. But when doing a broadcast left semi join, the left_chunk is reused many times, one per chunk, which I think means it can't be moved.

[TomAugspurger]

I included some changes to our cpp/scripts/ndsh.py to make testing this easier

python cpp/scripts/ndsh.py run-and-validate --input-dir scale-1 --output-dir validation  --benchmark-dir "cpp/build/benchmarks/ndsh/" --benchmark-args='--no-pinned-host-memory' --reuse-expected -q 4