Manually Construct Multi-GPU C++ TPC-H Queries

[quasiben]

Using a DGXH100 at SF3K, build, test, validate, and benchmark the following tpc-h queries using C++ RapidsMPF:

• Q1
• Q3
• Q4
• Q9
• Q13
• Q17
• Q18
• Q21

With core functionality now baked into rapidsmpf, these queries should be able to be implemented without "significant" development. They are representative of small, middle, large (complex) queries and would serve as a measuring stick for how well rapidsmpf streaming is performing at medium scale workloads. Additionally, this can guide mg-polars work as it progresses

[beckernick]

Would suggest flagging any scenarios in which manually constructed queries materially differ from what we'd expect to get from a robust query planner/optimizer (e.g., if we make assumptions in the implementation that would not be plausible if the plan came from DuckDB).

Ideally there would be nothing to flag :)

[TomAugspurger]

I'll take a look at query 4 in https://github.com/TomAugspurger/rapidsmpf/tree/tom/streaming-q4.

[quasiben]

mpiexec can be challenging to use, I've found using @pentschev's recent rrun additon to rapidsmpf to be helpful in getting early multit-gpu testing done (though I don't think it is fully setup for optimal numa-gpu coordination)

Here's an example:

LIBCUDF_NUM_HOST_WORKERS=4 KVIKIO_NTHREADS=4 UCX_TLS=^sm UCX_MAX_RNDV_RAILS=1 ./cpp/build/tools/rrun -n 1 ./cpp/build/benchmarks/ndsh/q9 --input-directory /path/to/scale-3000/ --output-file q9-sf3k.pq --num-streaming-threads 4 --num-rows-per-chunk 100000 --use-shuffle-join

[wence-]

I have q3 in flight and will take a look at q21

[nirandaperera]

I'll take Q13

[TomAugspurger]

I have an implementation for query 4 in the last couple of commits at main...TomAugspurger:rapidsmpf:tom/streaming-q4 that works at small scales.

I'm still working through how best to do the semi join. cudf's current filtered join implementation only supports using the right table as the build table, and the left table as the probe table. In Q4, orders is the left table and is smaller, so this isn't ideal. AFAICT, we won't be able to reuse the build table across chunks: we'll need to hash partition the larger lineitem table and construct the cudf::filtered_join on each chunk. But I'm still working through all this.

Otherwise, the only notable thing is that I'm not currently pushing the filters down into the read parquet. I think we'd need some of the enhancements from rapidsai/cudf#17142 to support the conditions we use in this query:

• column_a < column_b
• scalar_a < column < scalar_b (AKA (scalar_a < column) & (column < scalar_b))

[rjzamora]

I can take a look at q18 if no one is already working on it.