Handle MERGE WHEN NOT MATCHED BY SOURCE on hypertables

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

@@ -0,0 +1 @@
+Fixes: #9986 Handle MERGE WHEN NOT MATCHED BY SOURCE on hypertables

src/nodes/modify_hypertable_exec.c

@@ -2431,15 +2431,26 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
 		if (TupIsNull(context.planSlot))
 			break;
 
-		if (operation == CMD_INSERT || (operation == CMD_MERGE && (node->mt_merge_subcommands & MERGE_INSERT)))
+		/*
+		 * Only route tuples that get inserted: plain INSERTs and MERGE tuples
+		 * not matched by target (those have a NULL row identity). Routing a
+		 * matched or NOT MATCHED BY SOURCE tuple projects the INSERT over a
+		 * NULL source, putting a NULL into the partitioning column.
+		 */
+		bool route_insert = operation == CMD_INSERT;
+		if (operation == CMD_MERGE && (node->mt_merge_subcommands & MERGE_INSERT))
+		{
+			bool isNull = true;
+			if (AttributeNumberIsValid(resultRelInfo->ri_RowIdAttNo))
+				ExecGetJunkAttribute(context.planSlot, resultRelInfo->ri_RowIdAttNo, &isNull);
+			route_insert = isNull;
+		}
+
+		if (route_insert)
 		{
 			TupleTableSlot *hypertable_slot = context.planSlot;
 			if (operation == CMD_MERGE)
 			{
-				/*
-				 * XXX do we need an additional support of NOT MATCHED BY SOURCE
-				 * for PG >= 17? See PostgreSQL commit 0294df2f1f84
-				 */
 #if PG17_GE
 				List *actionStates = ctr->root_rri->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET];
 #else
@@ -2555,12 +2566,6 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
 				continue;
 			}
 
-			/*
-			 * copy INSERT merge action list to result relation info of corresponding chunk
-			 *
-			 * XXX do we need an additional support of NOT MATCHED BY SOURCE
-			 * for PG >= 17? See PostgreSQL commit 0294df2f1f84
-			 */
 			if (operation == CMD_MERGE)
 #if PG17_GE
 				ctr->cis->result_relation_info->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET] =
@@ -2982,16 +2987,21 @@ ExecMergeMatched(ModifyTableContext *context, ResultRelInfo *resultRelInfo, Item
 	bool isNull;
 	EPQState *epqstate = &mtstate->mt_epqstate;
 	ListCell *l;
+#if PG17_GE
+	List *actionStates;
+#endif
 
 	TupleTableSlot *rslot = NULL;
 
 	Assert(*matched == true);
 
 	/*
-	 * If there are no WHEN MATCHED actions, we are done.
+	 * If there are no WHEN MATCHED or WHEN NOT MATCHED BY SOURCE actions, we
+	 * are done.
 	 */
 #if PG17_GE
-	if (resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED] == NIL)
+	if (resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED] == NIL &&
+		resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_SOURCE] == NIL)
 		return NULL;
 #else
 	if (resultRelInfo->ri_matchedMergeAction == NIL)
@@ -3033,7 +3043,12 @@ lmerge_matched:;
 										   resultRelInfo->ri_oldTupleSlot))
 		elog(ERROR, "failed to fetch the target tuple");
 #if PG17_GE
-	foreach (l, resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED])
+	if (ExecQual(resultRelInfo->ri_MergeJoinCondition, econtext))
+		actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_MATCHED];
+	else
+		actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_SOURCE];
+
+	foreach (l, actionStates)
 	{
 #else
 	foreach (l, resultRelInfo->ri_matchedMergeAction)
@@ -3446,9 +3461,6 @@ ExecMergeNotMatched(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
 	 *
 	 * XXX does this mean that we can avoid creating copies of
 	 * actionStates on partitioned tables, for not-matched actions?
-	 *
-	 * XXX do we need an additional support of NOT MATCHED BY SOURCE
-	 * for PG >= 17? See PostgreSQL commit 0294df2f1f84
 	 */
 #if PG17_GE
 	actionStates = resultRelInfo->ri_MergeActions[MERGE_WHEN_NOT_MATCHED_BY_TARGET];

test/expected/merge_not_matched_by_source.out

@@ -0,0 +1,134 @@
+-- This file and its contents are licensed under the Apache License 2.0.
+-- Please see the included NOTICE for copyright information and
+-- LICENSE-APACHE for a copy of the license.
+-- Tests for MERGE ... WHEN NOT MATCHED BY SOURCE on hypertables (PG17+).
+-- A NOT MATCHED BY SOURCE action is only run by TimescaleDB when the MERGE is
+-- routed through ModifyHypertable, which happens when it also has an INSERT
+-- (NOT MATCHED BY TARGET) action. Each MERGE is run against both a plain table
+-- and a hypertable and the results are compared.
+-- Exact reproduction from the issue: the NOT MATCHED BY SOURCE DELETE must
+-- remove the unmatched target row instead of being silently skipped.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '100 days');
+   create_hypertable   
+-----------------------
+ (1,public,merge_ht,t)
+
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-01', 2, 20);
+insert into merge_ht select * from merge_plain;
+create table merge_src (k int, v int);
+insert into merge_src values (1, 100), (3, 300);
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
+when not matched by source then delete;
+select k, v from merge_ht order by k;
+ k |  v  
+---+-----
+ 1 |  10
+ 3 | 300
+
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+ result 
+--------
+ same
+
+drop table merge_plain, merge_ht, merge_src;
+-- All three match kinds together, with a conditional NOT MATCHED BY SOURCE,
+-- spanning multiple chunks.
+create table target_pg (time timestamptz not null, location int, temperature int);
+create table target_ht (time timestamptz not null, location int, temperature int);
+select create_hypertable('target_ht', 'time', chunk_time_interval => interval '1 day');
+   create_hypertable    
+------------------------
+ (2,public,target_ht,t)
+
+insert into target_pg
+select time, location, 14
+from generate_series('2021-01-01'::timestamptz, '2021-01-04', interval '1 day') time,
+     generate_series(1, 4) location;
+insert into target_ht select * from target_pg;
+-- Source matches some target rows, misses others, and adds a new one.
+create table source (location int, temperature int);
+insert into source values (1, 80), (2, 80), (5, 80);
+merge into target_pg t using source s on t.location = s.location
+when matched then update set temperature = (t.temperature + s.temperature) / 2
+when not matched by target then insert (time, location, temperature)
+     values ('2021-01-01', s.location, s.temperature)
+when not matched by source and t.location = 3 then delete
+when not matched by source then update set temperature = 99;
+merge into target_ht t using source s on t.location = s.location
+when matched then update set temperature = (t.temperature + s.temperature) / 2
+when not matched by target then insert (time, location, temperature)
+     values ('2021-01-01', s.location, s.temperature)
+when not matched by source and t.location = 3 then delete
+when not matched by source then update set temperature = 99;
+select case when exists (table target_ht except table target_pg)
+              or exists (table target_pg except table target_ht)
+            then 'different' else 'same' end as result;
+ result 
+--------
+ same
+
+drop table target_pg, target_ht, source;
+-- INSERT takes the partitioning column from a source column.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
+   create_hypertable   
+-----------------------
+ (3,public,merge_ht,t)
+
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
+insert into merge_ht select * from merge_plain;
+create table merge_src (ts timestamptz, k int, v int);
+insert into merge_src values ('2024-02-01', 1, 100), ('2024-02-05', 9, 900);
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+ result 
+--------
+ same
+
+drop table merge_plain, merge_ht, merge_src;
+-- Empty source: every target row is NOT MATCHED BY SOURCE.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
+   create_hypertable   
+-----------------------
+ (4,public,merge_ht,t)
+
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
+insert into merge_ht select * from merge_plain;
+create table merge_src (ts timestamptz, k int, v int);
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+select count(*) as merge_ht_rows from merge_ht;
+ merge_ht_rows 
+---------------
+             0
+
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+ result 
+--------
+ same
+
+drop table merge_plain, merge_ht, merge_src;

test/sql/CMakeLists.txt

@@ -130,7 +130,8 @@ if(CMAKE_BUILD_TYPE MATCHES Debug)
 endif(CMAKE_BUILD_TYPE MATCHES Debug)
 
 if((${PG_VERSION_MAJOR} GREATER_EQUAL "17"))
-  list(APPEND TEST_FILES tableam_alter_defaults.sql)
+  list(APPEND TEST_FILES tableam_alter_defaults.sql
+       merge_not_matched_by_source.sql)
 endif()
 
 if((${PG_VERSION_MAJOR} GREATER_EQUAL "18"))

test/sql/merge_not_matched_by_source.sql

@@ -0,0 +1,114 @@
+-- This file and its contents are licensed under the Apache License 2.0.
+-- Please see the included NOTICE for copyright information and
+-- LICENSE-APACHE for a copy of the license.
+
+-- Tests for MERGE ... WHEN NOT MATCHED BY SOURCE on hypertables (PG17+).
+-- A NOT MATCHED BY SOURCE action is only run by TimescaleDB when the MERGE is
+-- routed through ModifyHypertable, which happens when it also has an INSERT
+-- (NOT MATCHED BY TARGET) action. Each MERGE is run against both a plain table
+-- and a hypertable and the results are compared.
+
+-- Exact reproduction from the issue: the NOT MATCHED BY SOURCE DELETE must
+-- remove the unmatched target row instead of being silently skipped.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '100 days');
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-01', 2, 20);
+insert into merge_ht select * from merge_plain;
+
+create table merge_src (k int, v int);
+insert into merge_src values (1, 100), (3, 300);
+
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values ('2024-01-01', s.k, s.v)
+when not matched by source then delete;
+
+select k, v from merge_ht order by k;
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+
+drop table merge_plain, merge_ht, merge_src;
+
+-- All three match kinds together, with a conditional NOT MATCHED BY SOURCE,
+-- spanning multiple chunks.
+create table target_pg (time timestamptz not null, location int, temperature int);
+create table target_ht (time timestamptz not null, location int, temperature int);
+select create_hypertable('target_ht', 'time', chunk_time_interval => interval '1 day');
+insert into target_pg
+select time, location, 14
+from generate_series('2021-01-01'::timestamptz, '2021-01-04', interval '1 day') time,
+     generate_series(1, 4) location;
+insert into target_ht select * from target_pg;
+
+-- Source matches some target rows, misses others, and adds a new one.
+create table source (location int, temperature int);
+insert into source values (1, 80), (2, 80), (5, 80);
+
+merge into target_pg t using source s on t.location = s.location
+when matched then update set temperature = (t.temperature + s.temperature) / 2
+when not matched by target then insert (time, location, temperature)
+     values ('2021-01-01', s.location, s.temperature)
+when not matched by source and t.location = 3 then delete
+when not matched by source then update set temperature = 99;
+merge into target_ht t using source s on t.location = s.location
+when matched then update set temperature = (t.temperature + s.temperature) / 2
+when not matched by target then insert (time, location, temperature)
+     values ('2021-01-01', s.location, s.temperature)
+when not matched by source and t.location = 3 then delete
+when not matched by source then update set temperature = 99;
+
+select case when exists (table target_ht except table target_pg)
+              or exists (table target_pg except table target_ht)
+            then 'different' else 'same' end as result;
+
+drop table target_pg, target_ht, source;
+
+-- INSERT takes the partitioning column from a source column.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
+insert into merge_ht select * from merge_plain;
+
+create table merge_src (ts timestamptz, k int, v int);
+insert into merge_src values ('2024-02-01', 1, 100), ('2024-02-05', 9, 900);
+
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+
+drop table merge_plain, merge_ht, merge_src;
+
+-- Empty source: every target row is NOT MATCHED BY SOURCE.
+create table merge_plain (ts timestamptz not null, k int, v int);
+create table merge_ht (ts timestamptz not null, k int, v int);
+select create_hypertable('merge_ht', 'ts', chunk_time_interval => interval '1 day');
+insert into merge_plain values ('2024-01-01', 1, 10), ('2024-01-02', 2, 20), ('2024-01-03', 3, 30);
+insert into merge_ht select * from merge_plain;
+
+create table merge_src (ts timestamptz, k int, v int);
+
+merge into merge_plain t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+merge into merge_ht t using merge_src s on t.k = s.k
+when not matched by target then insert (ts, k, v) values (s.ts, s.k, s.v)
+when not matched by source then delete;
+
+select count(*) as merge_ht_rows from merge_ht;
+select case when exists (table merge_ht except table merge_plain)
+              or exists (table merge_plain except table merge_ht)
+            then 'different' else 'same' end as result;
+
+drop table merge_plain, merge_ht, merge_src;