Skip to content

Commit 6d26881

Browse files
svenklemmsvenklemm
committed
Handle MERGE WHEN NOT MATCHED BY SOURCE on hypertables
A MERGE on a hypertable that combined a WHEN NOT MATCHED BY TARGET insert with a WHEN NOT MATCHED BY SOURCE action silently skipped the NOT MATCHED BY SOURCE part. Target rows that had no matching source row were never deleted or updated. Fixes #9974
1 parent 02f78eb commit 6d26881

5 files changed

Lines changed: 280 additions & 18 deletions

File tree

‎.unreleased/merge-not-matched-by-source‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Fixes: #9986 Handle MERGE WHEN NOT MATCHED BY SOURCE on hypertables

‎src/nodes/modify_hypertable_exec.c‎

Lines changed: 29 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -2431,15 +2431,26 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
24312431
if (TupIsNull(context.planSlot))
24322432
break;
24332433

2434-
if (operation == CMD_INSERT || (operation == CMD_MERGE && (node->mt_merge_subcommands & MERGE_INSERT)))
2434+
/*
2435+
* Only route tuples that get inserted: plain INSERTs and MERGE tuples
2436+
* not matched by target (those have a NULL row identity). Routing a
2437+
* matched or NOT MATCHED BY SOURCE tuple projects the INSERT over a
2438+
* NULL source, putting a NULL into the partitioning column.
2439+
*/
2440+
bool route_insert = operation == CMD_INSERT;
2441+
if (operation == CMD_MERGE && (node->mt_merge_subcommands & MERGE_INSERT))
2442+
{
2443+
bool isNull = true;
2444+
if (AttributeNumberIsValid(resultRelInfo->ri_RowIdAttNo))
2445+
ExecGetJunkAttribute(context.planSlot, resultRelInfo->ri_RowIdAttNo, &isNull);
2446+
route_insert = isNull;
2447+
}
2448+
2449+
if (route_insert)
24352450
{
24362451
TupleTableSlot *hypertable_slot = context.planSlot;
24372452
if (operation == CMD_MERGE)
24382453
{
2439-
/*
2440-
* XXX do we need an additional support of NOT MATCHED BY SOURCE
2441-
* for PG >= 17? See PostgreSQL commit 0294df2f1f84
2442-
*/
24432454
#if PG17_GE
24442455
List *actionStates = ctr->root_rri->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET];
24452456
#else
@@ -2555,12 +2566,6 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
25552566
continue;
25562567
}
25572568

2558-
/*
2559-
* copy INSERT merge action list to result relation info of corresponding chunk
2560-
*
2561-
* XXX do we need an additional support of NOT MATCHED BY SOURCE
2562-
* for PG >= 17? See PostgreSQL commit 0294df2f1f84
2563-
*/
25642569
if (operation == CMD_MERGE)
25652570
#if PG17_GE
25662571
ctr->cis->result_relation_info->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET] =
@@ -2982,16 +2987,21 @@ ExecMergeMatched(ModifyTableContext *context, ResultRelInfo *resultRelInfo, Item
29822987
bool isNull;
29832988
EPQState *epqstate = &mtstate->mt_epqstate;
29842989
ListCell *l;
2990+
#if PG17_GE
2991+
List *actionStates;
2992+
#endif
29852993

29862994
TupleTableSlot *rslot = NULL;
29872995

29882996
Assert(*matched == true);
29892997

29902998
/*
2991-
* If there are no WHEN MATCHED actions, we are done.
2999+
* If there are no WHEN MATCHED or WHEN NOT MATCHED BY SOURCE actions, we
3000+
* are done.
29923001
*/
29933002
#if PG17_GE
2994-
if (resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED] == NIL)
3003+
if (resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED] == NIL &&
3004+
resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_SOURCE] == NIL)
29953005
return NULL;
29963006
#else
29973007
if (resultRelInfo->ri_matchedMergeAction == NIL)
@@ -3033,7 +3043,12 @@ lmerge_matched:;
30333043
resultRelInfo->ri_oldTupleSlot))
30343044
elog(ERROR, "failed to fetch the target tuple");
30353045
#if PG17_GE
3036-
foreach (l, resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED])
3046+
if (ExecQual(resultRelInfo->ri_MergeJoinCondition, econtext))
3047+
actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED];
3048+
else
3049+
actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_SOURCE];
3050+
3051+
foreach (l, actionStates)
30373052
{
30383053
#else
30393054
foreach (l, resultRelInfo->ri_matchedMergeAction)
@@ -3446,9 +3461,6 @@ ExecMergeNotMatched(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
34463461
*
34473462
* XXX does this mean that we can avoid creating copies of
34483463
* actionStates on partitioned tables, for not-matched actions?
3449-
*
3450-
* XXX do we need an additional support of NOT MATCHED BY SOURCE
3451-
* for PG >= 17? See PostgreSQL commit 0294df2f1f84
34523464
*/
34533465
#if PG17_GE
34543466
actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET];

‎test/expected/merge_not_matched_by_source.out‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
-- This file and its contents are licensed under the Apache License 2.0.
2+
-- Please see the included NOTICE for copyright information and
3+
-- LICENSE-APACHE for a copy of the license.
4+
-- Tests for MERGE ... WHEN NOT MATCHED BY SOURCE on hypertables (PG17+).
5+
-- A NOT MATCHED BY SOURCE action is only run by TimescaleDB when the MERGE is
6+
-- routed through ModifyHypertable, which happens when it also has an INSERT
7+
-- (NOT MATCHED BY TARGET) action. Each MERGE is run against both a plain table
8+
-- and a hypertable and the results are compared.
9+
-- Exact reproduction from the issue: the NOT MATCHED BY SOURCE DELETE must
10+
-- remove the unmatched target row instead of being silently skipped.
11+
create table merge_plain (ts timestamptz not null, k int, v int);
12+
create table merge_ht (ts timestamptz not null, k int, v int);
13+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '100 days');
14+
create_hypertable
15+
-----------------------
16+
(1,public,merge_ht,t)
17+
18+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-01', 2, 20);
19+
insert into merge_ht select * from merge_plain;
20+
create table merge_src (k int, v int);
21+
insert into merge_src values (1, 100), (3, 300);
22+
merge into merge_plain t using merge_src s on t.k = s.k
23+
when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
24+
when not matched by source then delete;
25+
merge into merge_ht t using merge_src s on t.k = s.k
26+
when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
27+
when not matched by source then delete;
28+
select k, v from merge_ht order by k;
29+
k | v
30+
---+-----
31+
1 | 10
32+
3 | 300
33+
34+
select case when exists (table merge_ht except table merge_plain)
35+
or exists (table merge_plain except table merge_ht)
36+
then 'different' else 'same' end as result;
37+
result
38+
--------
39+
same
40+
41+
drop table merge_plain, merge_ht, merge_src;
42+
-- All three match kinds together, with a conditional NOT MATCHED BY SOURCE,
43+
-- spanning multiple chunks.
44+
create table target_pg (time timestamptz not null, location int, temperature int);
45+
create table target_ht (time timestamptz not null, location int, temperature int);
46+
select create_hypertable('target_ht', 'time', chunk_time_interval => interval '1 day');
47+
create_hypertable
48+
------------------------
49+
(2,public,target_ht,t)
50+
51+
insert into target_pg
52+
select time, location, 14
53+
from generate_series('2021-01-01'::timestamptz, '2021-01-04', interval '1 day') time,
54+
generate_series(1, 4) location;
55+
insert into target_ht select * from target_pg;
56+
-- Source matches some target rows, misses others, and adds a new one.
57+
create table source (location int, temperature int);
58+
insert into source values (1, 80), (2, 80), (5, 80);
59+
merge into target_pg t using source s on t.location = s.location
60+
when matched then update set temperature = (t.temperature + s.temperature) / 2
61+
when not matched by target then insert (time, location, temperature)
62+
values ('2021-01-01', s.location, s.temperature)
63+
when not matched by source and t.location = 3 then delete
64+
when not matched by source then update set temperature = 99;
65+
merge into target_ht t using source s on t.location = s.location
66+
when matched then update set temperature = (t.temperature + s.temperature) / 2
67+
when not matched by target then insert (time, location, temperature)
68+
values ('2021-01-01', s.location, s.temperature)
69+
when not matched by source and t.location = 3 then delete
70+
when not matched by source then update set temperature = 99;
71+
select case when exists (table target_ht except table target_pg)
72+
or exists (table target_pg except table target_ht)
73+
then 'different' else 'same' end as result;
74+
result
75+
--------
76+
same
77+
78+
drop table target_pg, target_ht, source;
79+
-- INSERT takes the partitioning column from a source column.
80+
create table merge_plain (ts timestamptz not null, k int, v int);
81+
create table merge_ht (ts timestamptz not null, k int, v int);
82+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
83+
create_hypertable
84+
-----------------------
85+
(3,public,merge_ht,t)
86+
87+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
88+
insert into merge_ht select * from merge_plain;
89+
create table merge_src (ts timestamptz, k int, v int);
90+
insert into merge_src values ('2024-02-01', 1, 100), ('2024-02-05', 9, 900);
91+
merge into merge_plain t using merge_src s on t.k = s.k
92+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
93+
when not matched by source then delete;
94+
merge into merge_ht t using merge_src s on t.k = s.k
95+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
96+
when not matched by source then delete;
97+
select case when exists (table merge_ht except table merge_plain)
98+
or exists (table merge_plain except table merge_ht)
99+
then 'different' else 'same' end as result;
100+
result
101+
--------
102+
same
103+
104+
drop table merge_plain, merge_ht, merge_src;
105+
-- Empty source: every target row is NOT MATCHED BY SOURCE.
106+
create table merge_plain (ts timestamptz not null, k int, v int);
107+
create table merge_ht (ts timestamptz not null, k int, v int);
108+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
109+
create_hypertable
110+
-----------------------
111+
(4,public,merge_ht,t)
112+
113+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
114+
insert into merge_ht select * from merge_plain;
115+
create table merge_src (ts timestamptz, k int, v int);
116+
merge into merge_plain t using merge_src s on t.k = s.k
117+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
118+
when not matched by source then delete;
119+
merge into merge_ht t using merge_src s on t.k = s.k
120+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
121+
when not matched by source then delete;
122+
select count(*) as merge_ht_rows from merge_ht;
123+
merge_ht_rows
124+
---------------
125+
0
126+
127+
select case when exists (table merge_ht except table merge_plain)
128+
or exists (table merge_plain except table merge_ht)
129+
then 'different' else 'same' end as result;
130+
result
131+
--------
132+
same
133+
134+
drop table merge_plain, merge_ht, merge_src;

‎test/sql/CMakeLists.txt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,8 @@ if(CMAKE_BUILD_TYPE MATCHES Debug)
130130
endif(CMAKE_BUILD_TYPE MATCHES Debug)
131131

132132
if((${PG_VERSION_MAJOR} GREATER_EQUAL "17"))
133-
list(APPEND TEST_FILES tableam_alter_defaults.sql)
133+
list(APPEND TEST_FILES tableam_alter_defaults.sql
134+
merge_not_matched_by_source.sql)
134135
endif()
135136

136137
if((${PG_VERSION_MAJOR} GREATER_EQUAL "18"))

‎test/sql/merge_not_matched_by_source.sql‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
-- This file and its contents are licensed under the Apache License 2.0.
2+
-- Please see the included NOTICE for copyright information and
3+
-- LICENSE-APACHE for a copy of the license.
4+
5+
-- Tests for MERGE ... WHEN NOT MATCHED BY SOURCE on hypertables (PG17+).
6+
-- A NOT MATCHED BY SOURCE action is only run by TimescaleDB when the MERGE is
7+
-- routed through ModifyHypertable, which happens when it also has an INSERT
8+
-- (NOT MATCHED BY TARGET) action. Each MERGE is run against both a plain table
9+
-- and a hypertable and the results are compared.
10+
11+
-- Exact reproduction from the issue: the NOT MATCHED BY SOURCE DELETE must
12+
-- remove the unmatched target row instead of being silently skipped.
13+
create table merge_plain (ts timestamptz not null, k int, v int);
14+
create table merge_ht (ts timestamptz not null, k int, v int);
15+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '100 days');
16+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-01', 2, 20);
17+
insert into merge_ht select * from merge_plain;
18+
19+
create table merge_src (k int, v int);
20+
insert into merge_src values (1, 100), (3, 300);
21+
22+
merge into merge_plain t using merge_src s on t.k = s.k
23+
when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
24+
when not matched by source then delete;
25+
merge into merge_ht t using merge_src s on t.k = s.k
26+
when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
27+
when not matched by source then delete;
28+
29+
select k, v from merge_ht order by k;
30+
select case when exists (table merge_ht except table merge_plain)
31+
or exists (table merge_plain except table merge_ht)
32+
then 'different' else 'same' end as result;
33+
34+
drop table merge_plain, merge_ht, merge_src;
35+
36+
-- All three match kinds together, with a conditional NOT MATCHED BY SOURCE,
37+
-- spanning multiple chunks.
38+
create table target_pg (time timestamptz not null, location int, temperature int);
39+
create table target_ht (time timestamptz not null, location int, temperature int);
40+
select create_hypertable('target_ht', 'time', chunk_time_interval => interval '1 day');
41+
insert into target_pg
42+
select time, location, 14
43+
from generate_series('2021-01-01'::timestamptz, '2021-01-04', interval '1 day') time,
44+
generate_series(1, 4) location;
45+
insert into target_ht select * from target_pg;
46+
47+
-- Source matches some target rows, misses others, and adds a new one.
48+
create table source (location int, temperature int);
49+
insert into source values (1, 80), (2, 80), (5, 80);
50+
51+
merge into target_pg t using source s on t.location = s.location
52+
when matched then update set temperature = (t.temperature + s.temperature) / 2
53+
when not matched by target then insert (time, location, temperature)
54+
values ('2021-01-01', s.location, s.temperature)
55+
when not matched by source and t.location = 3 then delete
56+
when not matched by source then update set temperature = 99;
57+
merge into target_ht t using source s on t.location = s.location
58+
when matched then update set temperature = (t.temperature + s.temperature) / 2
59+
when not matched by target then insert (time, location, temperature)
60+
values ('2021-01-01', s.location, s.temperature)
61+
when not matched by source and t.location = 3 then delete
62+
when not matched by source then update set temperature = 99;
63+
64+
select case when exists (table target_ht except table target_pg)
65+
or exists (table target_pg except table target_ht)
66+
then 'different' else 'same' end as result;
67+
68+
drop table target_pg, target_ht, source;
69+
70+
-- INSERT takes the partitioning column from a source column.
71+
create table merge_plain (ts timestamptz not null, k int, v int);
72+
create table merge_ht (ts timestamptz not null, k int, v int);
73+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
74+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
75+
insert into merge_ht select * from merge_plain;
76+
77+
create table merge_src (ts timestamptz, k int, v int);
78+
insert into merge_src values ('2024-02-01', 1, 100), ('2024-02-05', 9, 900);
79+
80+
merge into merge_plain t using merge_src s on t.k = s.k
81+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
82+
when not matched by source then delete;
83+
merge into merge_ht t using merge_src s on t.k = s.k
84+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
85+
when not matched by source then delete;
86+
87+
select case when exists (table merge_ht except table merge_plain)
88+
or exists (table merge_plain except table merge_ht)
89+
then 'different' else 'same' end as result;
90+
91+
drop table merge_plain, merge_ht, merge_src;
92+
93+
-- Empty source: every target row is NOT MATCHED BY SOURCE.
94+
create table merge_plain (ts timestamptz not null, k int, v int);
95+
create table merge_ht (ts timestamptz not null, k int, v int);
96+
select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
97+
insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
98+
insert into merge_ht select * from merge_plain;
99+
100+
create table merge_src (ts timestamptz, k int, v int);
101+
102+
merge into merge_plain t using merge_src s on t.k = s.k
103+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
104+
when not matched by source then delete;
105+
merge into merge_ht t using merge_src s on t.k = s.k
106+
when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
107+
when not matched by source then delete;
108+
109+
select count(*) as merge_ht_rows from merge_ht;
110+
select case when exists (table merge_ht except table merge_plain)
111+
or exists (table merge_plain except table merge_ht)
112+
then 'different' else 'same' end as result;
113+
114+
drop table merge_plain, merge_ht, merge_src;

0 commit comments

Comments
 (0)