ARROW2: Performance benchmark

[houqp]

Perform thorough testing to make sure the arrow2 branch will indeed result in better performance across the board.

Please give https://github.com/apache/arrow-datafusion/tree/arrow2 a try for your workloads and report any regression you find here.

[houqp]

Here are some of the TPCH results I got from running our tpch benchmark suite on an 8 cores x86-64 Linux machine.

The base commit from I used as baseline for arrow-rs is 2008b1d. The commit for arrow2 is c0c9c72.

Default single partition CSV files generated from our tpch gen script (--batch-size 4096):

CSV tables partitioned into 16 files and processed with 8 datafusion partitions (--batch-size 4096 --partitions 8):

Data generated with:

PARTITIONS=16
for i in `seq 1 ${PARTITIONS}`; do
  echo "generating partition $i"
  ./dbgen -vf -s 16 -C ${PARTITIONS} -S $i
  ls -lh
done

Parquet tables partitioned in 8 files and processed with 8 datafusion partitions (--batch-size 4096 --partitions 8):

Note query 7 could not complete with arrow2 due to OOM. Arrow2 parquet reader currently consuming almost double the memory compared to arrow-rs is a known issue. Related upstream issue: jorgecarleitao/arrow2#768.

Q1 is significantly slower in arrow2 compared to the other queies (perhaps related to predicate pushdown?).

I think both of these two regressions require deepdive before we merge arrow2 into master.

Overall, arrow2 is around 5% faster for CSV tables and 10-20% faster for parquet tables across the board.

[tustvold]

Big 👍 to this, getting some concrete numbers would be really nice.

FWIW some ideas for whoever picks this up that I at least would be very interested in:

• Performance against current arrow-rs master, a number of non-trivial performance improvements have landed in the last month, with more currently in progress

• Performance of floating point aggregates, I seem to remember all the TPCH queries testing such fail to run correctly, but I could be mistaken

• Performance of dictionary arrays, there is a substantial amount of work completed and in-flight to improve this situation as it has historically been poor (and still is WIP)- Implement dyn scalar comparison kernels arrow-rs#1113, Switch datafusion to using eq_dyn_scalar, etc kernels #1610, Use dynamic arrow compute kernels  #1474, Preserve dictionary encoding when decoding parquet into Arrow arrays, 60x perf improvement (#171) arrow-rs#1180, etc...

• Performance on parquet files with reasonable row group sizes, the OOM would suggest they aren't teeny but wanted to clarify - there is currently a limitation of arrow-rs's parquet writer that makes it produce impractically small row groups - Write Multiple RecordBatch to Parquet Row Group arrow-rs#1211

• Performance of specific operators, e.g. FilterExec or SortPreservingMerge or ParquetExec, I'd basically be interested in where the performance gains are, and where we might gain or lose performance in a potential switch. Perhaps some execution plan metrics, or a perf dump or something 🤔

[alamb]

A number of the performance improvements will be in arrow-rs 8.0.0, though some such as apache/arrow-rs#1180 will not be released until arrow-rs 9.0.0

[houqp]

bummer that the dictionary encoding optimization will need to wait for the 9.0 release, but we can test it with a 9.0 datafusion branch if needed.

[sunchao]

I think a big part of perf improvement comes from Parquet predicate pushdown (e.g., stats, dictionary, bloom filter and column index). Does either arrow-rs or arrow2 implement (some of) these currently?

[alamb]

DataFusion does implement row group pruning based on statistics (that arrow-rs creates)

arrow-rs creates statistics. It will (in version after 8.0.0) preserve dictionary encoding (meaning the output of a dictionary encoded column will also be dictionary encoded) which should be a large win for low cardinality columns

It does not currently create or use column indexes or bloom filters

I believe that @tustvold has a plan for implementing more sophisticated predicate pushdown (aka that a filter on one column could be used to avoid decoding swaths of others) but I am not sure what the timeline on that is -- apache/arrow-rs#1191

[sunchao]

It will (in version after 8.0.0) preserve dictionary encoding (meaning the output of a dictionary encoded column will also be dictionary encoded) which should be a large win for low cardinality columns

This is nice 👍 . By dictionary-based row group filtering I mean we can first decode only the dictionary pages for the column chunks where pushed down predicates (e.g., in predicate) are associated to, and skip the whole row group if no keys in the dictionary match.

[alamb]

This is nice 👍 . By dictionary-based row group filtering I mean we can first decode only the dictionary pages for the column chunks where pushed down predicates (e.g., in predicate) are associated to, and skip the whole row group if no keys in the dictionary match.

this would definitely be a neat optimization

[tustvold]

skip the whole row group if no keys in the dictionary match.

You'd need to check that column chunk hadn't spilled, but yes that is broadly speaking the access pattern I wish to support with the work in apache/arrow-rs#1191. Specially exploit apache/arrow-rs#1180 to cheaply get dictionary data, evaluate predicates on it, and use this to construct a refined row selection to fetch.

As @alamb alludes to the timelines for this effort are not clear, and there may be other more pressing things for me to spend time on, but it is definitely something that I hope to deliver in the next month or so.

[jorgecarleitao]

• arrow-rs: group filter push down
• arrow2: group filter push down, page filter push down

afaik both support reading and writing dictionary encoded arrays to dictionary-encoded column chunks (but neither supports pushdown based on dict values atm).

TBH, imo the biggest limiting factor in implementing anything in parquet is its lack of documentation - it is just a lot of work to decipher what is being requested, and the situation is not improving. For example, I spent a lot of time in understanding the encodings, have a PR to try to help future folks implementing it, and it has been lingering for ~9 months now. I wish parquet was a bit more inspired by e.g. Arrow or ORC in this respect.

Note that datafusion's benchmarks only use "required" / non-nullable data, so most optimizations on the null values are not seen from datafusion's perspective. Last time I benched arrow2/parquet2 was much faster in nullable data; I am a bit surprised to see so many differences in non-nullable data.

[alamb]

arrow-rs 8 has some improvements for nullable columns as well now courtesy of @tustvold apache/arrow-rs#1054