Fix EvalPlanQual rechecking during MERGE.

Under some circumstances, concurrent MERGE operations could lead to
inconsistent results, that varied according the plan chosen. This was
caused by a lack of rowmarks on the source relation, which meant that
EvalPlanQual rechecking was not guaranteed to return the same source
tuples when re-running the join query.

Fix by ensuring that preprocess_rowmarks() sets up PlanRowMarks for
all non-target relations used in MERGE, in the same way that it does
for UPDATE and DELETE.

Per bug #18103. Back-patch to v15, where MERGE was introduced.

Dean Rasheed, reviewed by Richard Guo.

Discussion: https://postgr.es/m/18103-c4386baab8e355e3%40postgresql.org

5 files changed, 221 insertions, 23 deletions

diff --git a/src/test/isolation/expected/merge-join.out b/src/test/isolation/expected/merge-join.out
new file mode 100644
index 00000000000..57f048c52e5
--- /dev/null
+++ b/src/test/isolation/expected/merge-join.out
@@ -0,0 +1,148 @@
+Parsed test spec with 2 sessions
+
+starting permutation: b1 m1 s1 c1 b2 m2 s2 c2
+step b1: BEGIN ISOLATION LEVEL READ COMMITTED;
+step m1: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+step s1: SELECT * FROM tgt;
+id|val
+--+---
+ 1| 10
+ 2| 20
+ 3| 30
+(3 rows)
+
+step c1: COMMIT;
+step b2: BEGIN ISOLATION LEVEL READ COMMITTED;
+step m2: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+step s2: SELECT * FROM tgt;
+id|val
+--+---
+ 1| 10
+ 2| 20
+ 3| 30
+(3 rows)
+
+step c2: COMMIT;
+
+starting permutation: b1 b2 m1 hj ex m2 c1 c2 s1
+step b1: BEGIN ISOLATION LEVEL READ COMMITTED;
+step b2: BEGIN ISOLATION LEVEL READ COMMITTED;
+step m1: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+step hj: SET LOCAL enable_mergejoin = off; SET LOCAL enable_nestloop = off;
+step ex: EXPLAIN (verbose, costs off)
+           MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+QUERY PLAN                                         
+---------------------------------------------------
+Merge on public.tgt                                
+  ->  Hash Left Join                               
+        Output: tgt.ctid, src.val, src.id, src.ctid
+        Inner Unique: true                         
+        Hash Cond: (src.id = tgt.id)               
+        ->  Seq Scan on public.src                 
+              Output: src.val, src.id, src.ctid    
+        ->  Hash                                   
+              Output: tgt.ctid, tgt.id             
+              ->  Seq Scan on public.tgt           
+                    Output: tgt.ctid, tgt.id       
+(11 rows)
+
+step m2: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); <waiting ...>
+step c1: COMMIT;
+step m2: <... completed>
+step c2: COMMIT;
+step s1: SELECT * FROM tgt;
+id|val
+--+---
+ 1| 10
+ 2| 20
+ 3| 30
+(3 rows)
+
+
+starting permutation: b1 b2 m1 mj ex m2 c1 c2 s1
+step b1: BEGIN ISOLATION LEVEL READ COMMITTED;
+step b2: BEGIN ISOLATION LEVEL READ COMMITTED;
+step m1: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+step mj: SET LOCAL enable_hashjoin = off; SET LOCAL enable_nestloop = off;
+step ex: EXPLAIN (verbose, costs off)
+           MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+QUERY PLAN                                         
+---------------------------------------------------
+Merge on public.tgt                                
+  ->  Merge Left Join                              
+        Output: tgt.ctid, src.val, src.id, src.ctid
+        Inner Unique: true                         
+        Merge Cond: (src.id = tgt.id)              
+        ->  Index Scan using src_pkey on public.src
+              Output: src.val, src.id, src.ctid    
+        ->  Index Scan using tgt_pkey on public.tgt
+              Output: tgt.ctid, tgt.id             
+(9 rows)
+
+step m2: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); <waiting ...>
+step c1: COMMIT;
+step m2: <... completed>
+step c2: COMMIT;
+step s1: SELECT * FROM tgt;
+id|val
+--+---
+ 1| 10
+ 2| 20
+ 3| 30
+(3 rows)
+
+
+starting permutation: b1 b2 m1 nl ex m2 c1 c2 s1
+step b1: BEGIN ISOLATION LEVEL READ COMMITTED;
+step b2: BEGIN ISOLATION LEVEL READ COMMITTED;
+step m1: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+step nl: SET LOCAL enable_hashjoin = off; SET LOCAL enable_mergejoin = off;
+step ex: EXPLAIN (verbose, costs off)
+           MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val);
+QUERY PLAN                                         
+---------------------------------------------------
+Merge on public.tgt                                
+  ->  Nested Loop Left Join                        
+        Output: tgt.ctid, src.val, src.id, src.ctid
+        Inner Unique: true                         
+        ->  Seq Scan on public.src                 
+              Output: src.val, src.id, src.ctid    
+        ->  Index Scan using tgt_pkey on public.tgt
+              Output: tgt.ctid, tgt.id             
+              Index Cond: (tgt.id = src.id)        
+(9 rows)
+
+step m2: MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); <waiting ...>
+step c1: COMMIT;
+step m2: <... completed>
+step c2: COMMIT;
+step s1: SELECT * FROM tgt;
+id|val
+--+---
+ 1| 10
+ 2| 20
+ 3| 30
+(3 rows)
+
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 4fc56ae99c9..b2be88ead1d 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -51,6 +51,7 @@ test: merge-insert-update
 test: merge-delete
 test: merge-update
 test: merge-match-recheck
+test: merge-join
 test: delete-abort-savept
 test: delete-abort-savept-2
 test: aborted-keyrevoke
diff --git a/src/test/isolation/specs/merge-join.spec b/src/test/isolation/specs/merge-join.spec
new file mode 100644
index 00000000000..e33a02ccabc
--- /dev/null
+++ b/src/test/isolation/specs/merge-join.spec
@@ -0,0 +1,45 @@
+# MERGE JOIN
+#
+# This test checks the EPQ recheck mechanism during MERGE when joining to a
+# source table using different join methods, per bug #18103
+
+setup
+{
+  CREATE TABLE src (id int PRIMARY KEY, val int);
+  CREATE TABLE tgt (id int PRIMARY KEY, val int);
+  INSERT INTO src SELECT x, x*10 FROM generate_series(1,3) g(x);
+  INSERT INTO tgt SELECT x, x FROM generate_series(1,3) g(x);
+}
+
+teardown
+{
+  DROP TABLE src, tgt;
+}
+
+session s1
+step b1  { BEGIN ISOLATION LEVEL READ COMMITTED; }
+step m1  { MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); }
+step s1  { SELECT * FROM tgt; }
+step c1  { COMMIT; }
+
+session s2
+step b2  { BEGIN ISOLATION LEVEL READ COMMITTED; }
+step hj  { SET LOCAL enable_mergejoin = off; SET LOCAL enable_nestloop = off; }
+step mj  { SET LOCAL enable_hashjoin = off; SET LOCAL enable_nestloop = off; }
+step nl  { SET LOCAL enable_hashjoin = off; SET LOCAL enable_mergejoin = off; }
+step ex  { EXPLAIN (verbose, costs off)
+           MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); }
+step m2  { MERGE INTO tgt USING src ON tgt.id = src.id
+             WHEN MATCHED THEN UPDATE SET val = src.val
+             WHEN NOT MATCHED THEN INSERT VALUES (src.id, src.val); }
+step s2  { SELECT * FROM tgt; }
+step c2  { COMMIT; }
+
+permutation b1 m1 s1 c1 b2 m2 s2 c2
+permutation b1 b2 m1 hj ex m2 c1 c2 s1
+permutation b1 b2 m1 mj ex m2 c1 c2 s1
+permutation b1 b2 m1 nl ex m2 c1 c2 s1
diff --git a/src/test/regress/expected/merge.out b/src/test/regress/expected/merge.out
index 133d42117c0..28a6d0ba98b 100644
--- a/src/test/regress/expected/merge.out
+++ b/src/test/regress/expected/merge.out
@@ -1829,11 +1829,11 @@ MERGE INTO pa_target t USING pa_source s ON t.tid = s.sid
  Merge on public.pa_target t
    Merge on public.pa_targetp t_1
    ->  Hash Left Join
-         Output: s.sid, t_1.tableoid, t_1.ctid
+         Output: s.sid, s.ctid, t_1.tableoid, t_1.ctid
          Inner Unique: true
          Hash Cond: (s.sid = t_1.tid)
          ->  Seq Scan on public.pa_source s
-               Output: s.sid
+               Output: s.sid, s.ctid
          ->  Hash
                Output: t_1.tid, t_1.tableoid, t_1.ctid
                ->  Seq Scan on public.pa_targetp t_1
@@ -1859,11 +1859,11 @@ MERGE INTO pa_target t USING pa_source s ON t.tid = s.sid
 --------------------------------------------
  Merge on public.pa_target t
    ->  Hash Left Join
-         Output: s.sid, t.ctid
+         Output: s.sid, s.ctid, t.ctid
          Inner Unique: true
          Hash Cond: (s.sid = t.tid)
          ->  Seq Scan on public.pa_source s
-               Output: s.sid
+               Output: s.sid, s.ctid
          ->  Hash
                Output: t.tid, t.ctid
                ->  Result
diff --git a/src/test/regress/expected/with.out b/src/test/regress/expected/with.out
index 88e57a2c877..a01efa50a51 100644
--- a/src/test/regress/expected/with.out
+++ b/src/test/regress/expected/with.out
@@ -3015,28 +3015,30 @@ WITH cte_basic AS MATERIALIZED (SELECT 1 a, 'cte_basic val' b)
 MERGE INTO m USING (select 0 k, 'merge source SubPlan' v offset 0) o ON m.k=o.k
 WHEN MATCHED THEN UPDATE SET v = (SELECT b || ' merge update' FROM cte_basic WHERE cte_basic.a = m.k LIMIT 1)
 WHEN NOT MATCHED THEN INSERT VALUES(o.k, o.v);
-                           QUERY PLAN                           
-----------------------------------------------------------------
+                            QUERY PLAN                             
+-------------------------------------------------------------------
  Merge on public.m
    CTE cte_basic
      ->  Result
            Output: 1, 'cte_basic val'::text
    ->  Hash Right Join
-         Output: m.ctid, (0), ('merge source SubPlan'::text)
-         Hash Cond: (m.k = (0))
+         Output: m.ctid, o.k, o.v, o.*
+         Hash Cond: (m.k = o.k)
          ->  Seq Scan on public.m
                Output: m.ctid, m.k
          ->  Hash
-               Output: (0), ('merge source SubPlan'::text)
-               ->  Result
-                     Output: 0, 'merge source SubPlan'::text
+               Output: o.k, o.v, o.*
+               ->  Subquery Scan on o
+                     Output: o.k, o.v, o.*
+                     ->  Result
+                           Output: 0, 'merge source SubPlan'::text
    SubPlan 2
      ->  Limit
            Output: ((cte_basic.b || ' merge update'::text))
            ->  CTE Scan on cte_basic
                  Output: (cte_basic.b || ' merge update'::text)
                  Filter: (cte_basic.a = m.k)
-(19 rows)
+(21 rows)
 
 -- InitPlan
 WITH cte_init AS MATERIALIZED (SELECT 1 a, 'cte_init val' b)
@@ -3056,8 +3058,8 @@ WITH cte_init AS MATERIALIZED (SELECT 1 a, 'cte_init val' b)
 MERGE INTO m USING (select 1 k, 'merge source InitPlan' v offset 0) o ON m.k=o.k
 WHEN MATCHED THEN UPDATE SET v = (SELECT b || ' merge update' FROM cte_init WHERE a = 1 LIMIT 1)
 WHEN NOT MATCHED THEN INSERT VALUES(o.k, o.v);
-                          QUERY PLAN                           
----------------------------------------------------------------
+                             QUERY PLAN                             
+--------------------------------------------------------------------
  Merge on public.m
    CTE cte_init
      ->  Result
@@ -3069,15 +3071,17 @@ WHEN NOT MATCHED THEN INSERT VALUES(o.k, o.v);
                  Output: (cte_init.b || ' merge update'::text)
                  Filter: (cte_init.a = 1)
    ->  Hash Right Join
-         Output: m.ctid, (1), ('merge source InitPlan'::text)
-         Hash Cond: (m.k = (1))
+         Output: m.ctid, o.k, o.v, o.*
+         Hash Cond: (m.k = o.k)
          ->  Seq Scan on public.m
                Output: m.ctid, m.k
          ->  Hash
-               Output: (1), ('merge source InitPlan'::text)
-               ->  Result
-                     Output: 1, 'merge source InitPlan'::text
-(19 rows)
+               Output: o.k, o.v, o.*
+               ->  Subquery Scan on o
+                     Output: o.k, o.v, o.*
+                     ->  Result
+                           Output: 1, 'merge source InitPlan'::text
+(21 rows)
 
 -- MERGE source comes from CTE:
 WITH merge_source_cte AS MATERIALIZED (SELECT 15 a, 'merge_source_cte val' b)
@@ -3111,14 +3115,14 @@ WHEN NOT MATCHED THEN INSERT VALUES(o.a, o.b || (SELECT merge_source_cte.*::text
      ->  CTE Scan on merge_source_cte merge_source_cte_2
            Output: ((merge_source_cte_2.*)::text || ' merge insert'::text)
    ->  Hash Right Join
-         Output: m.ctid, merge_source_cte.a, merge_source_cte.b
+         Output: m.ctid, merge_source_cte.a, merge_source_cte.b, merge_source_cte.*
          Hash Cond: (m.k = merge_source_cte.a)
          ->  Seq Scan on public.m
                Output: m.ctid, m.k
          ->  Hash
-               Output: merge_source_cte.a, merge_source_cte.b
+               Output: merge_source_cte.a, merge_source_cte.b, merge_source_cte.*
                ->  CTE Scan on merge_source_cte
-                     Output: merge_source_cte.a, merge_source_cte.b
+                     Output: merge_source_cte.a, merge_source_cte.b, merge_source_cte.*
 (20 rows)
 
 DROP TABLE m;