Re: hashjoins vs. Bloom filters (yet again)

Hi,

Here's a significantly improved version of the v2 patch, using Bloom
filters in a single hashjoin node (no pushdown).

I'm still 100% convinced the goal should be filter pushdown - that's
where the real gains are - saving CPU time (cheaper Bloom probes), disk
I/O (by discarding tuples early with nbatch>1) and join order robustness
(join order less vulnerable/sensitive to misestimates).

I've been thinking about how to move this patch series forward, in that
direction, in an incremental way. No matter how we get to do a pushdown
(whether the bottom-up or opportunistic approach, discussed elsewhere in
this thread), it'll still need to figure out all the questions as the
local filters. Say, sizing the filter, populating the filter (in serial
and parallel cases), reporting stuff in explain, and so on. So it seems
reasonable to introduce that first, even if the gains are much smaller,
and then build the pushdown stuff on that, somehow.

v2.1
----

I spent the last couple weeks working on this, rebasing and cleaning up
the v2 patch, fixing all kinds of bugs in the PoC patch. The attached
v2.1 is much cleaner, maybe ~90% to committable (but as we all know, the
remaining 10% may take another 90% of the development time). There's
still a couple of FIXMEs and XXX with open questions, but I'm not aware
of any "actual" bugs.

I planned to also rework the pushdown patches on top of it, but I have
not managed to do that yet - the cleanup took me longer, and I have to
take a break from the patch for a bit. I'm not sure when/if I'll have
time for that again. So I'm posting at least the reworked v2.1, for
others to take a look.

The overall structure of the patch was basically right (easy to see by
comparing v2 and v2.1 branches). It's probably not worth going over all
the bugs I found in the v2 patch in detail, but it's what you'd expect
from a PoC patch:

- parallel queries probing filter too late, not before spilling them to
disk (efficiency)

- insufficient locking in parallel query (correctness)

- confusion when the GUC changes during query execution (weird, but can
happen)

- incorrect instrumentation in plans calling shutdown (spi, cursors)

- various improvements in the EXPLAIN (may need more work)

- incorrect accounting for probe hits/misses for joins with duplicates
in the inner relation (which meant incorrect build decisions)

- various other smaller issues

Where appropriate, I tried to briefly explain the issue in comments
before the fixed code - by explaining why it needs to be this way,
without being too verbose. If some changes are unclear, let me know and
I'll try to remember what the issue was, and why it's fixed this way.

I'm still a bit unsure about is the handling of instrumentation in
ExecShutdownHash. It took me ages to get it working correctly (I think),
so that it does not lose/discard instrumentation with SPI/cursors, etc.
The adaptive heuristics (building/enabling/disabling filters) relies on
that, so some queries were silently not using good filters etc.

filter sizing
-------------

The other "big" question is how to size the filter. There are two main
challenges, I think.

1) in some cases the filter is not really faster than the hash table

One thing I realized after looking at perf results is that even a very
selective filter (discarding 90%+ rows) may not be any faster than
checking the hash table. I believe this simply happens for hash tables
with very few buckets, because then the buckets are pretty much the same
as a Bloom filter.

Let's say we have a hash table with 128K buckets - each bucket is a
HashJoinTuple pointer, so this means 1MB of data (the whole hash table
may be much larger, but that does not matter). We size the buckets for
load factor 1.0 (NTUP_PER_BUCKET=1), so in most cases it's enough to
check if the bucket is NULL to reject the tuple. The Bloom filter won't
really save anything in this case, it may even be more expensive.

I think the solution will be to consider the size of the buckets array,
and not build the filter if the array is small. But what is small? It
probably depends on some L2/L3 cache size, but we'll need to pick some
sort of threshold. 1MB, maybe? We could also make it relative to the
filter size - if the filter is more than 1/8 of the buckets, it may not
help very much. Probably requires some microbenchmarks, to find what
values might work. Maybe if we had a way to read the L2/L3 sizes from
the system, that'd help us too.

FWIW this only applies to nbatch=1 case, of course. In batched joins the
savings come mainly from discarding tuples early, before spilling them
to disk. May need some thresholds too, but certainly much higher.

2) we don't really know the number of elements

We can build the filter in two places. While building the hash table,
and during probing. And in both cases we don't really know the number of
filter elements, which affects the sizing.

While building the hash table it's obvious - we only have an estimate of
the row count, but it may be inaccurate. If we decide to build the
filter half-way through the hash table build (when going from nbatch=1
to nbatch=2), we at least know the number of tuples seen so far, but at
this point we also know the estimate was wrong. The code size the filter
for 2x the tuples, but that's wishful thinking.

I wonder if we might do something smarter, like writing the hash values
into a file, and then buildign the filter from that. Yes, it will cost
something, but the hashvalues are tiny - much smaller than the tuples.
We could probably also deduplicate them cheaply. So maybe it's worth a
properly sized filter. Worth a try / experiment, I guess?

Another detail is that sizing the filter for tuples is not quite right.
We really care about number of distinct keys, and if the inner relation
has duplicates, it may be a much smaller value. Maybe it'd be worth
building a HLL counter, or something?

we'd only need to do any of this in the nbatch>1 case, ofc. With a
single batch we have all the values in the hashtable, which means we
know the exect row count, and we could maybe calculate distincts too.

benchmarks
----------

I'm attaching scripts + results from a simple benchmark on v2.1, joining
a "fact" table to one or more dimensions, varying the selectivity of the
join (by placing condition on the dimension, i.e. the inner relation).

The PDFs summarize the results from two machines, a recent ryzen 5 and
an old xeon. There are some differences, but the overall message is
mostly the same.

In both cases the most selective filter (keeping only 10% rows) wins
only for work_mem=1MB, because it's batched. It's not very visible The
single-batch runs got only mildly faster (by ~10%), and only with many
buckets (not very visible in the PDFs, check the raw CSV results).

I think this illustrates my earlier point about sizing the filter in
relation to the buckets array.

regards

--
Tomas Vondra