ARROW-11053: [Rust] [DataFusion] Optimize joins with dynamic capacity for output batches

[andygrove]

This builds on #9035.

I am investigating why join performance is so bad with smaller batch sizes (see https://issues.apache.org/jira/browse/ARROW-11030) and this is one optimization that I have found so far that helps a bit.

Prior to this PR, we use the size of left or right batches to guess the capacity of output batches and this results in a lot of over allocation in some cases. For TPC-H q12 at SF=100, I see vectors created with capacity of ~3,000,000 (the size of the build-side of the join) and then we only populate it with ~700 entries.

This PR attempts to learn a good capacity based on previously processed batches.

Here are query times in seconds at different batch sizes:

Batch Size	Master	This PR
16384	189.6	158.0
32768	61.9	47.2
65536	28.2	21.4
131072	19.0	15.6

[github-actions]

https://issues.apache.org/jira/browse/ARROW-11053

[Dandandan]

In #9032 it was benchmarked to be a bit faster currently to initialize with capacity 0 than to use the correct capacity upfront. I think that is more something curious and hopefully will change, but that might be the case here too? What is the performance if we initialize with 0 everywhere?

[Dandandan]

An important source of slowness seems to be in the (use and inefficiency of) creating the ArrayData and MutableArrayData structure. In profiling I see a lot in build_extend, freeze etc.

Changing the piece of code to generate a Vec<&ArrayData> directly gives a ~20% speedup locally on batches of size 1000 on your branch @andygrove :

        let (is_primary, arrays) = match primary[0].schema().index_of(field.name()) {
            Ok(i) => Ok((true, primary.iter().map(|batch| batch.column(i).data_ref().as_ref()).collect::<Vec<_>>())),
            Err(_) => {
                match secondary[0].schema().index_of(field.name()) {
                    Ok(i) => Ok((false, secondary.iter().map(|batch| batch.column(i).data_ref().as_ref()).collect::<Vec<_>>())),
                    _ => Err(DataFusionError::Internal(
                        format!("During execution, the column {} was not found in neither the left or right side of the join", field.name()).to_string()
                    ))
                }
            }
        }.map_err(DataFusionError::into_arrow_external_error)?;

[Dandandan]

Opened a PR with that change here: #9042

[Dandandan]

I think part of a further speed up could be moving the building of the left / build-side Vec<&ArrayData> arrays so that it is only created once instead of for each right batch in build_batch_from_indices. Currently when making the batch size smaller, the build-side Vec is built more times, but also contains more (smaller) batches itself, which could explain (part of the) big / exponential slowdown on smaller batches.

[Dandandan]

I think this is something we should merge. We can maybe tweak the extra 1024 capacity but this is one extra source of slowness for smaller batches. Also I think helps making the other sources more clear.

[Dandandan]

@andygrove thought more about this, I think we are able to use indices.len() for the exact required capacity rather than using previous sizes. I included the change among other changes in this PR #9048

[andygrove]

Closed in favor of #9048