ARROW-16713: [C++] Pull join accumulation outside of HashJoinImpl

[save-buffer]

No description provided.

[github-actions]

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

[michalursa]

Looks good to me.

I have only a few comments:

1. I would like ProbeBatches to be implemented outside of HashJoinImpl by calling ProbeSingleBatch method of HashJoinImpl. That way the implementation of ProbeBatches can be shared between both current HashJoinImpl and SwissJoin's implementation of it.
2. finished_mutex_ in hash_join.cc is probably not used anymore and could be removed.
3. I think I would prefer if AccumulationQueue was thread-safe than having HashJoinNode managing mutexes for accumulation queues. I see that in this implementation AccumulationQueue is sometimes used on a single-thread or in a read-only way, where the mutex is not necessary, but I would still prefer to extract a vector of exec batches in a thread-safe way from AccumulationQueue and use it instead in such situations. Or alternatively having a state flag in AccumulationQueue saying that it is read-only and doesn't require a mutex.

[save-buffer]

1. That's true. I did think it would be cleaner to have batches that are being operated on be owned by the data structure that's doing the work. For ProbeBatches specifically I could make a special "probing accumulation queue" that gets moved into instead.
2. Good point, will remove
3. I initially did have it be thread-safe, but then I realized that most of them would have locks managed externally anyway: for spilling we'll be using PartitionLocks and for HashJoinNode we use the mutexes to protect stuff other than the AccumulationQueues, so in neither case will we actually ever use the AccumulationQueue's mutex.

[michalursa]

Re AccumulationQueue: then it becomes just a vector, and maybe doesn't need promoting it to a separate class in a separate file

[westonpace]

Looks good. A few minor nits and some questions but otherwise I think this helps clear things up.

[westonpace]

Re AccumulationQueue: then it becomes just a vector, and maybe doesn't need promoting it to a separate class in a separate file

This is true. The only difference I see between the accumulation queue and a vector of batches at the moment is row_count() but, on second examination, it doesn't seem we are currently using row_count() anywhere.

[save-buffer]

row_count is used for the Bloom filter build (in only one spot). I kind of like AccumulationQueue because it also disallows copying, and the Append function is handy. Also in the next PR I'll be adding a SpillingAccumulationQueue, so I kind of like having a name for these. I'm fine getting rid of it though too though

[westonpace]

A few minor nits but this seems ready. Thanks for doing this, it's a good cleanup, above and beyond it's potential utility for spillover. I find this easier to follow than the old implementation.

[westonpace]

It would be nice to get some consistency with how we use int64_t, uint64_t, and size_t within the engine. Do you have any convention suggestions?

[save-buffer]

I'd say we should probably be using size_t for most things that are non-negative. That guy on the mailing list had an epic struggle building for 32-bit because we were using int64_t where we should've used size_t.
BTW here I use size_t because that's what std::vector::size() returns.

[save-buffer]

I think row_count can never be nagative either, so it should be a size_t. ExecBatch::length should probably be size_t as well.

[save-buffer]

The only failures seem to be with S3FS (unrelated to hash join completely) and the tracing span thingy which is being addressed in #13108

[westonpace]

I wish I knew why this PR triggers the span thing to flare up. It doesn't seem that any of the code you have is touching spans. I suspect it is something rather subtle. At the moment appveyor passes on other PRs so if I merge this it will break things, even if there is nothing in this PR that is related.

I've proposed a solution on #13108 that should work. So hopefully we can get that merged pretty quick. Otherwise we'll have to figure out what change here is causing the span bug to trigger.

[westonpace]

Now that the span issue is resolved it seems MSVC is happy so let's merge.