Assert failure in parallel query due to SubPlan double visit for hash join

Hi,

Whilst stress testing the next version of the stack-based
instrumentation patch [0], I ran into what appears to be an ancient
bug in how we walk plan state for SubPlan nodes in parallel query
setup and instrumentation handling.

Specifically, the planstate_tree_walker callbacks in execParallel.c
assume that a given PlanState is only walked once, but that doesn't
hold true for hash joins since (I think, not tested) Postgres 10. We
create two separate references to the same subplan PlanState, one via
hashclauses and the other via hashkeys, each having their own
SubPlanState, but going to the same subplan PlanState. And
planstate_walk_subplans() descends into the subtree once per
reference.

I believe this is an issue since b8d7f053c5c (Postgres 10, CCing
Andres as the author of that one) changed how ExecInitHashJoin calls
ExecInitExpr, creating the second SubPlanState. More digging also
found the very old a64846f3ada that talked about a seemingly related
problem, but that must have happened before parallel query was in the
picture.

We can see this cause an assertion failure on master with the
following reproducer:

CREATE TABLE big (a int, b int);
INSERT INTO big SELECT g, g FROM generate_series(1, 10000) g;
CREATE TABLE small (x int PRIMARY KEY, y int);
INSERT INTO small SELECT g, g FROM generate_series(1, 1000) g;
CREATE TABLE other (k int PRIMARY KEY, f int);
INSERT INTO other SELECT g, g FROM generate_series(1, 1000) g;
ANALYZE big, small, other;
ALTER TABLE other SET (parallel_workers = 0);

SET parallel_setup_cost = 0;
SET parallel_tuple_cost = 0;
SET min_parallel_table_scan_size = 0;
SET enable_nestloop = off;
SET enable_mergejoin = off;
SET enable_parallel_hash = off;
SET parallel_leader_participation = off; -- to show loop count more clearly

EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT COUNT(*) FROM big b JOIN small s ON s.x =
CASE WHEN (b.a, b.b) NOT IN (SELECT o.k, o.f FROM other o ORDER BY o.f)
THEN 1 ELSE 2 END;

TRAP: failed Assert("toc->toc_entry[i].key != key"), File:
"shm_toc.c", Line: 190

Besides the assertion failure it also causes problems for
instrumentation. In this example we get an incorrect loop count (the
subplan is really only run in the two workers, so the loops=16 below
should be 4), and the Sort node is missing the Sort Method reporting:

QUERY PLAN
--------------------------------------------------------------
Finalize Aggregate (actual rows=1.00 loops=1)
-> Gather (actual rows=2.00 loops=1)
Workers Planned: 2
Workers Launched: 2
-> Partial Aggregate (actual rows=1.00 loops=2)
-> Hash Join (actual rows=5000.00 loops=2)
Hash Cond: (CASE WHEN (NOT (ANY ((b.a = (hashed
SubPlan any_1).col1) AND (b.b = (hashed SubPlan any_1).col2)))) THEN 1
ELSE 2 END = s.x)
-> Parallel Seq Scan on big b (actual
rows=5000.00 loops=2)
-> Hash (actual rows=1000.00 loops=2)
Buckets: 1024 Batches: 1 Memory Usage: 44kB
-> Seq Scan on small s (actual
rows=1000.00 loops=2)
SubPlan any_1
-> Sort (actual rows=1000.00 loops=16)
Sort Key: o.f
-> Seq Scan on other o (actual
rows=1000.00 loops=16)

The assertion failure happens because ExecParallelInitializeDSM visits
that Sort twice and tries to shm_toc_insert() the same plan_node_id
key. The instrumentation issue happens because (1) for loops, we
accumulate the worker instrumentation twice in
ExecParallelReportInstrumentation and then again in
ExecParallelRetrieveInstrumentation for the nodes underneath the
shared SubPlan, (2) for "Sort Method" the leader and workers end up
pointing at different (duplicate) SharedSortInfo slots and the leader
looks at the empty one.

To fix this, the least convoluted way I could think of is to invent a
new mechanism to collect a deduplicated list of plan states, and then
call the existing logic on each list member. This also allows some
refactoring to remove logic for counting how many nodes we've visited.

With the attached patch, the instrumentation now shows correctly, and
the added regression test no longer fails with an assertion failure.

...
SubPlan any_1
-> Sort (actual rows=1000.00 loops=4)
Sort Key: o.f
Worker 0: Sort Method: quicksort Memory: 48kB
Worker 1: Sort Method: quicksort Memory: 48kB
-> Seq Scan on other o (actual
rows=1000.00 loops=4)

We could alternatively deal with this in the existing walkers, but
then we'd have to modify the walker context in four different places
to carry the identities of already visited plan states. I also assume
that changing how ExecInitHashJoin works is off the table, but I admit
that logic goes a bit over my head.

Note that technically the ExecShutdownNode_walker suffers from the
same problem, but it doesn't appear to be a practical problem with
in-core code today, and it's a bit more complicated to fix with the
current nested InstrStartNode assumptions (that I intend to fix
separately in the stack-based instrumentation series).

Thoughts?

Thanks,
Lukas

[0]: https://www.postgresql.org/message-id/flat/CAP53PkzdBK8VJ1fS4AZ481LgMN8f9mJiC39ZRHqkFUSYq6KWmg%40mail.gmail.com

--
Lukas Fittl