Re: [HACKERS] Parallel Hash take II

Hi,

On 2017-11-14 01:30:30 +1300, Thomas Munro wrote:
> > +-- The "good" case: batches required, but we plan the right number; we
> > +-- plan for 16 batches, and we stick to that number, and peak memory
> > +-- usage says within our work_mem budget
> > +-- non-parallel
> > +set max_parallel_workers_per_gather = 0;
> > +set work_mem = '128kB';
> >
> > So how do we know that's actually the case we're testing rather than
> > something arbitrarily different? There's IIRC tests somewhere that just
> > filter the json explain output to the right parts...
>
> Yeah, good idea. My earlier attempts to dump out the hash join
> dimensions ran into problems with architecture sensitivity and then
> some run-to-run non-determinism in the parallel case (due to varying
> fragmentation depending on how many workers get involved in time).
> The attached version tells you about batch growth without reporting
> the exact numbers, except in the "ugly" case where we know that there
> is only one possible outcome because the extreme skew detector is
> guaranteed to go off after the first nbatch increase (I got rid of all
> other tuples except ones with the same key to make this true).

Hm. The way you access this doesn't quite seem right:
+--
+-- exercises for the hash join code
+--
+begin;
+set min_parallel_table_scan_size = 0;
+set parallel_setup_cost = 0;
+-- Extract bucket and batch counts from an explain analyze plan. In
+-- general we can't make assertions about how many batches (or
+-- buckets) will be required because it can vary, but we can in some
+-- special cases and we can check for growth.
+create or replace function hash_join_batches(query text)
+returns table (original int, final int) language plpgsql
+as
+$$
+declare
+ line text;
+ matches text[];
+begin
+ for line in
+ execute 'explain analyze ' || query
+ loop
+ matches := (regexp_matches(line, ' Batches: ([0-9]+) \(originally ([0-9]+)\)'));
+ if matches is not null then
+ original := matches[2]::int;
+ final := matches[1]::int;
+ return next;
+ else
+ matches := regexp_matches(line, ' Batches: ([0-9]+)');
+ if matches is not null then
+ original := matches[1]::int;
+ final := original;
+ return next;
+ end if;
+ end if;
+ end loop;
+end;
+$$;

Why not use format json and access the output that way? Then you can be
sure you access the right part of the tree and such?

> > + else
> > + {
> > + errno = stat_errno;
> > + elog(LOG, "could not stat file \"%s\": %m", path);
> > + }
> >
> > All these messages are "not expected to ever happen" ones, right?
>
> You'd have to suffer a nasty filesystem failure, remount read-only or
> manually with permissions or something. Not sure where the line is,
> but I've changed all of these new elog calls to ereport.

Oh, I'd been fine keeping them as elogs. The one exception would have
been out-of-space cases which'll occur in practice.

> > + if (vfdP->fdstate & FD_TEMP_FILE_LIMIT)
> > + {
> > + /* Subtract its size from current usage (do first in case of error) */
> > + temporary_files_size -= vfdP->fileSize;
> > + vfdP->fileSize = 0;
> > + }
> >
> > So, is it right to do so unconditionally and without regard for errors?
> > If the file isn't deleted, it shouldn't be subtracted from fileSize. I
> > guess you're managing that through the flag, but that's not entirely
> > obvious.
>
> I think it is. Reasoning: The existing behaviour of fd.c is that if
> we don't manage to delete temporary files, we'll LOG something and
> forget about them (they'll be cleaned up eventually by a clean restart
> or human intervention).

IOW: Never ;)

> > +/*
> > + * Write a tuple. If a meta-data size was provided to sts_initialize, then a
> > + * pointer to meta data of that size must be provided.
> > + */
> > +void
> > +sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data,
> > + MinimalTuple tuple)
> > +{
> >
> > + /* Do we have space? */
> > + size = accessor->sts->meta_data_size + tuple->t_len;
> > + if (accessor->write_pointer + size >= accessor->write_end)
> > + {
> > + /* Try flushing to see if that creates enough space. */
> > + if (accessor->write_chunk != NULL)
> > + sts_flush_chunk(accessor);
> > +
> > + /*
> > + * It may still not be enough in the case of a gigantic tuple, or if
> > + * we haven't created a chunk buffer at all yet.
> > + */
> > + if (accessor->write_pointer + size >= accessor->write_end)
> > + {
> > + SharedTuplestoreParticipant *participant;
> > + size_t space_needed;
> > + int pages_needed;
> > +
> > + /* How many pages to hold this data and the chunk header? */
> > + space_needed = offsetof(SharedTuplestoreChunk, data) + size;
> > + pages_needed = (space_needed + (BLCKSZ - 1)) / BLCKSZ;
> > + pages_needed = Max(pages_needed, STS_MIN_CHUNK_PAGES);
> > +
> > + /*
> > + * Double the chunk size until it's big enough, and record that
> > + * fact in the shared expansion log so that readers know about it.
> > + */
> > + participant = &accessor->sts->participants[accessor->participant];
> > + while (accessor->write_pages < pages_needed)
> > + {
> > + accessor->write_pages *= 2;
> > + participant->chunk_expansion_log[participant->chunk_expansions++] =
> > + accessor->write_page;
> > + }
> >
> > Hm. Isn't that going to be pretty unfunny if you have one large and a
> > lot of small tuples?
>
> It will increase the parallel scan grain size, and then keep that size
> for the rest of the contents of one backend's output file. I am aware
> of two downsides to using a large parallel grain:

> 1. It determines the amount of unfairness when we run out of data:
> it's the maximum amount of extra data that the unlucky last worker can
> finish up with after all the others have finished. I think this
> effect is reduced by higher level factors: when a reader runs out of
> data in one backend's file, it'll start reading another backend's
> file. If it's hit the end of all backends' files and this is an outer
> batch, Parallel Hash will just go and work on another batch
> immediately.

Consider e.g. what happens if there's the occasional 500MB datum, and
the rest's very small...

> Better ideas?

Not really. I'm more than a bit suspicous of this solution, but I don't
really have a great suggestion otherwise. One way to combat extreme
size skew would be to put very large datums into different files.

But I think we probably can go with your approach for now, ignoring my
failure prone spidey senses ;)

> > + /* Find the location of a new chunk to read. */
> > + p = &accessor->sts->participants[accessor->read_participant];
> > +
> > + SpinLockAcquire(&p->mutex);
> > + eof = p->read_page >= p->npages;
> > + if (!eof)
> > + {
> > + /*
> > + * Figure out how big this chunk is. It will almost always be the
> > + * same as the last chunk loaded, but if there is one or more
> > + * entry in the chunk expansion log for this page then we know
> > + * that it doubled that number of times. This avoids the need to
> > + * do IO to adjust the read head, so we don't need to hold up
> > + * concurrent readers. (An alternative to this extremely rarely
> > + * run loop would be to use more space storing the new size in the
> > + * log so we'd have 'if' instead of 'while'.)
> > + */
> > + read_page = p->read_page;
> > + while (p->chunk_expansion < p->chunk_expansions &&
> > + p->chunk_expansion_log[p->chunk_expansion] == p->read_page)
> > + {
> > + p->chunk_pages *= 2;
> > + p->chunk_expansion++;
> > + }
> > + chunk_pages = p->chunk_pages;
> > +
> > + /* The next reader will start after this chunk. */
> > + p->read_page += chunk_pages;
> > + }
> > + SpinLockRelease(&p->mutex);
> >
> > This looks more like the job of an lwlock rather than a spinlock.
>
> I switched to the alternative algorithm mentioned in parentheses in
> the comment. It uses a bit more space, but that loop is gone. In my
> mind this is much like Parallel Seq Scan: we acquire a spinlock just
> to advance the block pointer. The added complication is that we also
> check if the chunk size has changed, which clang renders as this many
> instructions:
>
> postgres[0x10047eee0] <+176>: movslq 0x144(%r15,%rbx), %rcx
> postgres[0x10047eee8] <+184>: cmpl 0x140(%r15,%rbx), %ecx
> postgres[0x10047eef0] <+192>: jge 0x10047ef16 ;
> <+230> at sharedtuplestore.c:489
> postgres[0x10047eef2] <+194>: leaq (%r15,%rbx), %rdx
> postgres[0x10047eef6] <+198>: cmpl %r12d, 0x40(%rdx,%rcx,8)
> postgres[0x10047eefb] <+203>: jne 0x10047ef16 ;
> <+230> at sharedtuplestore.c:489
> postgres[0x10047eefd] <+205>: leaq 0x144(%r15,%rbx), %rsi
> postgres[0x10047ef05] <+213>: leal 0x1(%rcx), %edi
> postgres[0x10047ef08] <+216>: movl %edi, (%rsi)
> postgres[0x10047ef0a] <+218>: movl 0x44(%rdx,%rcx,8), %ecx
> postgres[0x10047ef0e] <+222>: movl %ecx, 0x148(%r15,%rbx)
> postgres[0x10047ef16] <+230>: movl 0x148(%r15,%rbx), %r15d
>
> That should be OK, right?

It's not too bad. Personally I'm of the opinion though that pretty much
no new spinlocks should be added - their worst case performance
characteristics are bad enough for that to be only worth the
experimentation in case swhere each cycle really matters and where
contention is unlikely.

> > One day we're going to need a better approach to this. I have no idea
> > how, but this per-node, and now per_node * max_parallelism, approach has
> > only implementation simplicity as its benefit.
>
> I agree, and I am interested in that subject. In the meantime, I
> think it'd be pretty unfair if parallel-oblivious hash join and
> sort-merge join and every other parallel plan get to use work_mem * p
> (and in some cases waste it with duplicate data), but Parallel Hash
> isn't allowed to do the same (and put it to good use).

I'm not sure I care about fairness between pieces of code ;)

> > - node->hj_JoinState = HJ_NEED_NEW_OUTER;
> > + if (hashtable->parallel_state)
> > + {
> > + Barrier *build_barrier;
> > +
> > + build_barrier = &hashtable->parallel_state->build_barrier;
> > + if (BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER)
> > + {
> > + /*
> > + * If multi-batch, we need to hash the outer relation
> > + * up front.
> > + */
> > + if (hashtable->nbatch > 1)
> > + ExecParallelHashJoinPartitionOuter(node);
> > + BarrierArriveAndWait(build_barrier,
> > + WAIT_EVENT_HASH_BUILD_HASHING_OUTER);
> > + }
> > + Assert(BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
> > +
> > + /* Each backend should now select a batch to work on. */
> > + hashtable->curbatch = -1;
> > + node->hj_JoinState = HJ_NEED_NEW_BATCH;
> > +
> > + continue;
> > + }
> > + else
> > + node->hj_JoinState = HJ_NEED_NEW_OUTER;
> >
> > You know what I'm going to say about all these branches, and sigh.
>
> BTW this is not per-tuple code -- it runs once at the end of hashing.
> Not sure what you're looking for here.

It was more a general statement about all the branches in nodeHashjoin,
than about these specific branches. Should've made that clearer. There's
definitely branches in very common parts:
case HJ_NEED_NEW_OUTER:

/*
* We don't have an outer tuple, try to get the next one
*/
if (hashtable->parallel_state)
outerTupleSlot =
ExecParallelHashJoinOuterGetTuple(outerNode, node,
&hashvalue);
else
outerTupleSlot =
ExecHashJoinOuterGetTuple(outerNode, node, &hashvalue);

I don't think you should do so now, but I think a reasonable approach
here would be to move the HJ_BUILD_HASHTABLE code into a separate
function (it really can't be hot). Then have specialized ExecHashJoin()
versions for parallel/non-parallel and potentially for outer/inner/anti.

> > If we don't split this into two versions, we at least should store
> > hashNode->parallel_state in a local var, so the compiler doesn't have to
> > pull that out of memory after every external function call (of which
> > there are a lot). In common cases it'll end up in a callee saved
> > registers, and most of the called functions won't be too register
> > starved (on x86-64).
>
> Hmm. Well I did that already in v24 -- in many places there is now a
> local variable called pstate.

See above piece of code, and a few others, in nodeHash.

> > I think it'd be better if we structured the file so we just sat guc's
> > with SET LOCAL inside a transaction.
>
> I wrapped the whole region of join.sql concerned with hash joins in a
> transaction that rolls back, so I don't have to write LOCAL. That's
> just as good, right?

Not really imo. Being able to read a test without going through all
previous ones is a lot better.

Greetings,

Andres Freund