Re: Improving scalability of Parallel Bitmap Heap/Index Scan

> Looking at the hash join memory calculation, it has to worry about
> different things (batching, skew etc), but it does have for the shared
> hash table:
>
> /*
> * Parallel Hash tries to use the combined hash_mem of all workers to
> * avoid the need to batch. If that won't work, it falls back to hash_mem
> * per worker and tries to process batches in parallel.
> */
> ...
> newlimit = (double) hash_table_bytes * (double) (parallel_workers + 1);
>
> ...so we can just do the same, although there's the question of the
> two copies: I think we may have to treat memory accounting differently
> before and after repartitioning. The latter can afford to be more
> pessimistic because of non-overlap. In the former stage, we have to
> worry about a page entry showing up in all workers' table, and we
> don't that to force our hands with lossification.

I changed the code such that the maximum TIDBitmap size during the index
scan phase is work_mem - work_mem / (max_participants + 1), and the
maximum TIDBitmap size during the partitioning phase is work_mem /
(max_participants + 1). This sums up to work_mem and takes care of the
two copies in the following way:

The split maximizes the memory we give to the index scan phase while
ensuring there's enough memory left to represent the same TIDBitmap in
its partitioned form during the partitioning phase.

During the index scan phase, each participant may encounter all matching
TIDs from the index. During the partitioning phase, each participant
only keeps the TIDs that hash to its partition, which is roughly
1/max_participants of the total TIDs.

Since the phases are sequential, each participant's total memory usage
stays within work_mem across both phases.

>>>> That's for sure less work than making TIDBitmap parallel-safe for insert>>>> (which includes lossification), should be on-par
performance-wise and
>>>> avoids the increase in memory consumption.
>>>
>>> That sounds difficult to review.
>>
>> A lock-free ring buffer is not that complicated and it would avoid the
>> increase in memory usage. At the same time, it would also disallow
>> splitting out the partitioning step into a separate plan node because
>> the "on-the-fly" partitioning must happen prior to creating the
>> TIDBitmap. I'll give this a try to see how involved it gets.
>
> I was trying to imply that I have zero interest in reviewing that.

Fine by me. We can start with the current approach and always improve on
it later.

> Before going further, I want to see what we're losing in the worst
> case of static partitioning: All pages happen to go to one worker
> after repartitioning.

With the hash-based partitioning scheme, the probability of all pages
ending up in a single participant depends on how many distinct 256-block
ranges the data spans. For K distinct ranges and N participants the
probability is (1/N)^(K-1), assuming a hash function which uniformly
distributes inputs over the key space. murmur hash has that property.
Some examples with 4 workers:

- 2 distinct ranges: (1/4)^1 = 25% chance they collide
- 3 distinct ranges: (1/4)^2 = 6.25% chance
- 10 distinct ranges: (1/4)^9 ≈ 0.0004% chance
- 100 distinct ranges: (1/4)^99 ≈ 10^-60% chance (practically zero)

The worst case can only realistically occur when the data spans few
block ranges, which are typically fast to process anyway, even with
single-threaded execution. Beyond that, when the planner expects the
index to only return little rows, Index Scan is typically costed cheaper
than BIS, so the planner avoids the BIS to begin with.

> So the next patch would be:
>
> 1. Simple and generous memory accounting for a start

Done.

> 2. Fix any bugs that would make performence testing a waste of time

Passes all tests, also on CI. I fixed one issue worth mentioning which
is mapping from participant number to a dense partition ID, e.g. when
the leader doesn't participate or workers are late so that they don't
participate and thereby leave a gap. Previously this resulted in
unassigned partitions.

There's a lengthy comment in the top of nodeBitmapIndexscan.c which
explains the partition ID computation, also across parallel BIS siblings.

> 3. Test the above worst case
>
> If that doesn't prove to be too bad compared to dynamic work
> portioning (i.e. does parallel index scan make up for the loss of
> parallel heap scan), we can worry about the other details.

See explanations above for why the worst-case with lots of data cannot
realistically happen. To create a test for the worstcase, I would have
to find block ranges for which murmurhash32() % nparticipants returns
the same value and then use a WHERE clause which only returns rows in
these block ranges. And that for a large number of block ranges.

Or did you have something else in mind here?

I tried testing the worst-case with little data where a single block
range ends up in the same participant. But with so little data I didn't
manage to trick the planner into choosing a parallel plan.

Beyond that there are the following changes:

1) I added the FastMod() function which avoids the costly modulo (=
division) operation for every single TID during partitioning. For
details see comments in the code.
2) Removed the sort_template.h optimization. It's shared in a separate
patch, see [1].
3) Drafted a proper commit message.

Last but not least, I realized that parallel BIS needs parallel
instrumentation support for EXPLAIN ANALYZE. I'll work on that next and
provide an updated patch the next days.

--
David Geier

[1]
https://www.postgresql.org/message-id/29f27b97-b3d7-476b-acae-832b3443b0c4%40gmail.com